Vasanko/sanic
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Add more documentationand type annotations

Browse Source
This commit is contained in:
Adam Hopkins 2021-01-31 12:30:37 +02:00
parent b958cdc151
commit 4358a7eefd
9 changed files with 396 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
docs/sanic/api_reference.rst
View File

@ -30,93 +30,109 @@ sanic.compat
.. automodule:: sanic.compat
:members:
:show-inheritance:
sanic.config
------------
.. automodule:: sanic.config
:members:
:show-inheritance:
sanic.cookies
-------------
.. automodule:: sanic.cookies
:members:
:show-inheritance:
sanic.errorpages
----------------
.. automodule:: sanic.errorpages
:members:
:show-inheritance:
sanic.exceptions
----------------
.. automodule:: sanic.exceptions
:members:
:show-inheritance:
sanic.handlers
--------------
.. automodule:: sanic.handlers
:members:
:show-inheritance:
sanic.http
----------
.. automodule:: sanic.http
:members:
:show-inheritance:
sanic.log
---------
.. automodule:: sanic.log
:members:
:show-inheritance:
sanic.request
-------------
.. automodule:: sanic.request
:members:
:show-inheritance:
sanic.response
--------------
.. automodule:: sanic.response
:members:
:show-inheritance:
sanic.router
------------
.. automodule:: sanic.router
:members:
:show-inheritance:
sanic.server
------------
.. automodule:: sanic.server
:members:
:show-inheritance:
sanic.static
------------
.. automodule:: sanic.static
:members:
:show-inheritance:
sanic.views
-----------
.. automodule:: sanic.views
:members:
:show-inheritance:
sanic.websocket
---------------
.. automodule:: sanic.websocket
:members:
:show-inheritance:
sanic.worker
------------
.. automodule:: sanic.worker
:members:
:show-inheritance:

5
sanic/app.py
View File

@ -21,6 +21,7 @@ from typing import (
Optional,
Set,
Type,
TypeVar,
Union,
)
from urllib.parse import urlencode, urlunparse
@ -85,7 +86,7 @@ class Sanic(
router: Router = None,
error_handler: ErrorHandler = None,
load_env: bool = True,
request_class: Request = None,
request_class: Type[Request] = None,
strict_slashes: bool = False,
log_config: Optional[Dict[str, Any]] = None,
configure_logging: bool = True,
@ -163,7 +164,7 @@ class Sanic(
also return a future, and the actual ensure_future call
is delayed until before server start.
`See user guide <https://sanicframework.org/guide/basics/tasks.html#background-tasks>`__
`See user guide <https://sanicframework.org/guide/basics/tasks.html#background-tasks>`_
:param task: future, couroutine or awaitable
"""

4
sanic/compat.py
View File

@ -14,13 +14,13 @@ class Header(CIMultiDict):
"""
Container used for both request and response headers. It is a subclass of
`CIMultiDict
<https://multidict.readthedocs.io/en/stable/multidict.html#cimultidictproxy>`__.
<https://multidict.readthedocs.io/en/stable/multidict.html#cimultidictproxy>`_.
It allows for multiple values for a single key in keeping with the HTTP
spec. Also, all keys are *case in-sensitive*.
Please checkout `the MultiDict documentation
<https://multidict.readthedocs.io/en/stable/multidict.html#multidict>`__
<https://multidict.readthedocs.io/en/stable/multidict.html#multidict>`_
for more details about how to use the object. In general, it should work
very similar to a regular dictionary.
"""

1
sanic/handlers.py
View File

@ -26,6 +26,7 @@ MiddlewareType = Union[RequestMiddlewareType, ResponseMiddlewareType]
ListenerType = Callable[
[Sanic, AbstractEventLoop], Optional[Coroutine[Any, Any, None]]
]
RouteHandler = Callable[..., Coroutine[Any, Any, HTTPResponse]]
class ErrorHandler:

3
sanic/headers.py
View File

@ -6,6 +6,9 @@ from urllib.parse import unquote
from sanic.helpers import STATUS_CODES
# TODO:
# - the Options object should be a typed object to allow for less casting
# across the application (in request.py for example)
HeaderIterable = Iterable[Tuple[str, Any]] # Values convertible to str
HeaderBytesIterable = Iterable[Tuple[bytes, bytes]]
Options = Dict[str, Union[int, str]] # key=value fields in various headers

129
sanic/request.py
View File

@ -1,22 +1,27 @@
from __future__ import annotations
from typing import (
Optional,
TYPE_CHECKING,
Any,
DefaultDict,
Dict,
List,
NamedTuple,
Optional,
Tuple,
Union,
)
if TYPE_CHECKING:
from sanic.server import ConnInfo
from sanic.app import Sanic
from sanic.http import Http
from asyncio.transports import BaseTransport
import email.utils
import uuid
from asyncio.transports import BaseTransport
from collections import defaultdict
from http.cookies import SimpleCookie
from types import SimpleNamespace
@ -27,6 +32,7 @@ from httptools import parse_url # type: ignore
from sanic.compat import CancelledErrors, Header
from sanic.exceptions import InvalidUsage
from sanic.headers import (
Options,
parse_content_header,
parse_forwarded,
parse_host,
@ -49,16 +55,21 @@ DEFAULT_HTTP_CONTENT_TYPE = "application/octet-stream"
class RequestParameters(dict):
"""Hosts a dict with lists as values where get returns the first
"""
Hosts a dict with lists as values where get returns the first
value of the list and getlist returns the whole shebang
"""
def get(self, name, default=None):
def get(self, name: str, default: Optional[Any] = None) -> Optional[Any]:
"""Return the first value, either the default or actual"""
return super().get(name, [default])[0]
def getlist(self, name, default=None):
"""Return the entire list"""
def getlist(
self, name: str, default: Optional[Any] = None
) -> Optional[Any]:
"""
Return the entire list
"""
return super().get(name, default)
@ -110,7 +121,7 @@ class Request:
self.raw_url = url_bytes
# TODO: Content-Encoding detection
self._parsed_url = parse_url(url_bytes)
self._id = None
self._id: Optional[Union[uuid.UUID, str, int]] = None
self.app = app
self.headers = headers
@ -123,7 +134,7 @@ class Request:
self.conn_info: Optional[ConnInfo] = None
self.ctx = SimpleNamespace()
self.name: Optional[str] = None
self.parsed_forwarded = None
self.parsed_forwarded: Optional[Options] = None
self.parsed_json = None
self.parsed_form = None
self.parsed_files = None
@ -136,7 +147,7 @@ class Request:
self.uri_template = None
self.request_middleware_started = False
self._cookies: Dict[str, str] = {}
self.stream = None
self.stream: Optional[Http] = None
self.endpoint = None
def __repr__(self):
@ -148,24 +159,30 @@ class Request:
return uuid.uuid4()
async def respond(
self, response=None, *, status=200, headers=None, content_type=None
self,
response: Optional[BaseHTTPResponse] = None,
*,
status: int = 200,
headers: Optional[Union[Header, Dict[str, str]]] = None,
content_type: Optional[str] = None,
):
# This logic of determining which response to use is subject to change
if response is None:
response = self.stream.response or HTTPResponse(
response = (self.stream and self.stream.response) or HTTPResponse(
status=status,
headers=headers,
content_type=content_type,
)
# Connect the response
if isinstance(response, BaseHTTPResponse):
if isinstance(response, BaseHTTPResponse) and self.stream:
response = self.stream.respond(response)
# Run response middleware
try:
response = await self.app._run_response_middleware(
self, response, request_name=self.name
)
except CancelledErrors:
# Redefining this as a tuple here satisfies mypy
except tuple(CancelledErrors):
raise
except Exception:
error_logger.exception(
@ -186,11 +203,35 @@ class Request:
self.body = b"".join([data async for data in self.stream])
@property
def id(self):
def id(self) -> Optional[Union[uuid.UUID, str, int]]:
"""
A request ID passed from the client, or generated from the backend.
By default, this will look in a request header defined at:
``self.app.config.REQUEST_ID_HEADER``. It defaults to
``X-Request-ID``. Sanic will try to cast the ID into a ``UUID`` or an
``int``. If there is not a UUID from the client, then Sanic will try
to generate an ID by calling ``Request.generate_id()``. The default
behavior is to generate a ``UUID``. You can customize this behavior
by subclassing ``Request``.
.. code-block:: python
from sanic import Request, Sanic
from itertools import count
class IntRequest(Request):
counter = count()
def generate_id(self):
return next(self.counter)
app = Sanic("MyApp", request_class=IntRequest)
"""
if not self._id:
self._id = self.headers.get(
self.app.config.REQUEST_ID_HEADER,
self.__class__.generate_id(self),
self.__class__.generate_id(self), # type: ignore
)
# Try casting to a UUID or an integer
@ -199,11 +240,11 @@ class Request:
self._id = uuid.UUID(self._id)
except ValueError:
try:
self._id = int(self._id)
self._id = int(self._id) # type: ignore
except ValueError:
...
return self._id
return self._id # type: ignore
@property
def json(self):
@ -378,9 +419,17 @@ class Request:
]
query_args = property(get_query_args)
"""
Convenience property to access :meth:`Request.get_query_args` with
default values.
"""
@property
def cookies(self) -> Dict[str, str]:
"""
:return: Incoming cookies on the request
:rtype: Dict[str, str]
"""
if self._cookies is None:
cookie = self.headers.get("Cookie")
if cookie is not None:
@ -432,13 +481,16 @@ class Request:
@property
def path(self) -> str:
"""Path of the local HTTP request."""
"""
:return: path of the local HTTP request
:rtype: str
"""
return self._parsed_url.path.decode("utf-8")
# Proxy properties (using SERVER_NAME/forwarded/request/transport info)
@property
def forwarded(self):
def forwarded(self) -> Options:
"""
Active proxy information obtained from request headers, as specified in
Sanic configuration.
@ -449,6 +501,9 @@ class Request:
- path is url-unencoded
Additional values may be available from new style Forwarded headers.
:return: forwarded address info
:rtype: Dict[str, str]
"""
if self.parsed_forwarded is None:
self.parsed_forwarded = (
@ -464,10 +519,14 @@ class Request:
Client IP address, if available.
1. proxied remote address `self.forwarded['for']`
2. local remote address `self.ip`
:return: IPv4, bracketed IPv6, UNIX socket name or arbitrary string
:rtype: str
"""
if not hasattr(self, "_remote_addr"):
self._remote_addr = self.forwarded.get("for", "") # or self.ip
self._remote_addr = str(
self.forwarded.get("for", "")
) # or self.ip
return self._remote_addr
@property
@ -477,12 +536,14 @@ class Request:
1. `config.SERVER_NAME` if in full URL format
2. proxied proto/scheme
3. local connection protocol
:return: http|https|ws|wss or arbitrary value given by the headers.
:rtype: str
"""
if "//" in self.app.config.get("SERVER_NAME", ""):
return self.app.config.SERVER_NAME.split("//")[0]
if "proto" in self.forwarded:
return self.forwarded["proto"]
return str(self.forwarded["proto"])
if (
self.app.websocket_enabled
@ -506,17 +567,20 @@ class Request:
3. request host header
hostname and port may be separated by
`sanic.headers.parse_host(request.host)`.
:return: the first matching host found, or empty string
:rtype: str
"""
server_name = self.app.config.get("SERVER_NAME")
if server_name:
return server_name.split("//", 1)[-1].split("/", 1)[0]
return self.forwarded.get("host") or self.headers.get("host", "")
return str(self.forwarded.get("host") or self.headers.get("host", ""))
@property
def server_name(self) -> str:
"""
The hostname the client connected to, by ``request.host``.
:return: hostname the client connected to, by ``request.host``
:rtype: str
"""
return parse_host(self.host)[0] or ""
@ -527,21 +591,26 @@ class Request:
``request.host``.
Default port is returned as 80 and 443 based on ``request.scheme``.
:return: port number
:rtype: int
"""
port = self.forwarded.get("port") or parse_host(self.host)[1]
return port or (80 if self.scheme in ("http", "ws") else 443)
return int(port or (80 if self.scheme in ("http", "ws") else 443))
@property
def server_path(self) -> str:
"""
Full path of current URL. Uses proxied or local path.
:return: full path of current URL; uses proxied or local path
:rtype: str
"""
return self.forwarded.get("path") or self.path
return str(self.forwarded.get("path") or self.path)
@property
def query_string(self) -> str:
"""
Representation of the requested query
:return: representation of the requested query
:rtype: str
"""
if self._parsed_url.query:
return self._parsed_url.query.decode("utf-8")
@ -551,7 +620,8 @@ class Request:
@property
def url(self) -> str:
"""
The URL
:return: the URL
:rtype: str
"""
return urlunparse(
(self.scheme, self.host, self.path, None, self.query_string, None)
@ -592,7 +662,8 @@ class Request:
class File(NamedTuple):
"""
Model for defining a file
Model for defining a file. It is a ``namedtuple``, therefore you can
iterate over the object, or access the parameters by name.
:param type: The mimetype, defaults to text/plain
:param body: Bytes of the file

260
sanic/response.py
View File

@ -1,10 +1,23 @@
from functools import partial
from mimetypes import guess_type
from os import path
from typing import Optional
from pathlib import PurePath
from typing import (
Any,
AnyStr,
Callable,
Coroutine,
Dict,
Iterator,
Optional,
Tuple,
Union,
)
from urllib.parse import quote_plus
from warnings import warn
from typing_extensions import Protocol
from sanic.compat import Header, open_async
from sanic.cookies import CookieJar
from sanic.helpers import has_message_body, remove_entity_headers
@ -21,29 +34,79 @@ except ImportError:
json_dumps = partial(dumps, separators=(",", ":"))
class HTMLProtocol(Protocol):
def __html__(self) -> AnyStr:
...
def _repr_html_(self) -> AnyStr:
...
class Range(Protocol):
def start(self) -> int:
...
def end(self) -> int:
...
def size(self) -> int:
...
def total(self) -> int:
...
class BaseHTTPResponse:
"""
The base class for all HTTP Responses
"""
def __init__(self):
self.asgi: bool = False
self.body: Optional[bytes] = None
self.content_type: Optional[str] = None
self.stream: Http = None
self.status: int = None
self.headers = Header({})
self._cookies: Optional[CookieJar] = None
def _encode_body(self, data):
def _encode_body(self, data: Optional[AnyStr]):
if data is None:
return b""
return data.encode() if hasattr(data, "encode") else data
return (
data.encode() if hasattr(data, "encode") else data # type: ignore
)
@property
def cookies(self):
def cookies(self) -> CookieJar:
"""
The response cookies. Cookies should be set and written as follows:
.. code-block:: python
response.cookies["test"] = "It worked!"
response.cookies["test"]["domain"] = ".yummy-yummy-cookie.com"
response.cookies["test"]["httponly"] = True
`See user guide
<https://sanicframework.org/guide/basics/cookies.html>`_
:return: the cookie jar
:rtype: CookieJar
"""
if self._cookies is None:
self._cookies = CookieJar(self.headers)
return self._cookies
@property
def processed_headers(self):
"""Obtain a list of header tuples encoded in bytes for sending.
def processed_headers(self) -> Iterator[Tuple[bytes, bytes]]:
"""
Obtain a list of header tuples encoded in bytes for sending.
Add and remove headers based on status and content_type.
:return: response headers
:rtype: Tuple[Tuple[bytes, bytes], ...]
"""
# TODO: Make a blacklist set of header names and then filter with that
if self.status in (304, 412): # Not Modified, Precondition Failed
@ -56,22 +119,66 @@ class BaseHTTPResponse:
for name, value in self.headers.items()
)
async def send(self, data=None, end_stream=None):
"""Send any pending response headers and the given data as body.
async def send(
self,
data: Optional[Union[AnyStr]] = None,
end_stream: Optional[bool] = None,
) -> None:
"""
Send any pending response headers and the given data as body.
:param data: str or bytes to be written
:end_stream: whether to close the stream after this block
:param end_stream: whether to close the stream after this block
"""
if data is None and end_stream is None:
end_stream = True
if end_stream and not data and self.stream.send is None:
return
data = data.encode() if hasattr(data, "encode") else data or b""
data = (
data.encode() # type: ignore
if hasattr(data, "encode")
else data or b""
)
await self.stream.send(data, end_stream=end_stream)
StreamingFunction = Callable[[BaseHTTPResponse], Coroutine[Any, Any, None]]
class StreamingHTTPResponse(BaseHTTPResponse):
"""Old style streaming response. Use `request.respond()` instead of this in
new code to avoid the callback."""
"""
Old style streaming response where you pass a streaming function:
.. code-block:: python
async def sample_streaming_fn(response):
await response.write("foo")
await asyncio.sleep(1)
await response.write("bar")
await asyncio.sleep(1)
@app.post("/")
async def test(request):
return stream(sample_streaming_fn)
.. warning::
**Deprecated** and set for removal in v21.6. You can now achieve the
same functionality without a callback.
.. code-block:: python
@app.post("/")
async def test(request):
response = await request.respond()
await response.send("foo", False)
await asyncio.sleep(1)
await response.send("bar", False)
await asyncio.sleep(1)
await response.send("", True)
return response
"""
__slots__ = (
"streaming_fn",
@ -83,10 +190,10 @@ class StreamingHTTPResponse(BaseHTTPResponse):
def __init__(
self,
streaming_fn,
status=200,
headers=None,
content_type="text/plain; charset=utf-8",
streaming_fn: StreamingFunction,
status: int = 200,
headers: Optional[Dict[str, str]] = None,
content_type: str = "text/plain; charset=utf-8",
chunked="deprecated",
):
if chunked != "deprecated":
@ -118,25 +225,40 @@ class StreamingHTTPResponse(BaseHTTPResponse):
class HTTPResponse(BaseHTTPResponse):
"""
HTTP response to be sent back to the client.
:param body: the body content to be returned
:type body: Optional[bytes]
:param status: HTTP response number. **Default=200**
:type status: int
:param headers: headers to be returned
:type headers: Optional;
:param content_type: content type to be returned (as a header)
:type content_type: Optional[str]
"""
__slots__ = ("body", "status", "content_type", "headers", "_cookies")
def __init__(
self,
body=None,
status=200,
headers=None,
content_type=None,
body: Optional[AnyStr] = None,
status: int = 200,
headers: Optional[Dict[str, str]] = None,
content_type: Optional[str] = None,
):
super().__init__()
self.content_type = content_type
self.content_type: Optional[str] = content_type
self.body = self._encode_body(body)
self.status = status
self.headers = Header(headers or {})
self._cookies = None
def empty(status=204, headers=None):
def empty(
status=204, headers: Optional[Dict[str, str]] = None
) -> HTTPResponse:
"""
Returns an empty response to the client.
@ -147,13 +269,13 @@ def empty(status=204, headers=None):
def json(
body,
status=200,
headers=None,
content_type="application/json",
dumps=json_dumps,
body: Any,
status: int = 200,
headers: Optional[Dict[str, str]] = None,
content_type: str = "application/json",
dumps: Callable[..., str] = json_dumps,
**kwargs,
):
) -> HTTPResponse:
"""
Returns response object with body in json format.
@ -171,8 +293,11 @@ def json(
def text(
body, status=200, headers=None, content_type="text/plain; charset=utf-8"
):
body: str,
status: int = 200,
headers: Optional[Dict[str, str]] = None,
content_type: str = "text/plain; charset=utf-8",
) -> HTTPResponse:
"""
Returns response object with body in text format.
@ -192,8 +317,11 @@ def text(
def raw(
body, status=200, headers=None, content_type="application/octet-stream"
):
body: Optional[AnyStr],
status: int = 200,
headers: Optional[Dict[str, str]] = None,
content_type: str = "application/octet-stream",
) -> HTTPResponse:
"""
Returns response object without encoding the body.
@ -210,7 +338,11 @@ def raw(
)
def html(body, status=200, headers=None):
def html(
body: Union[str, bytes, HTMLProtocol],
status: int = 200,
headers: Optional[Dict[str, str]] = None,
) -> HTTPResponse:
"""
Returns response object with body in html format.
@ -218,11 +350,13 @@ def html(body, status=200, headers=None):
:param status: Response code.
:param headers: Custom Headers.
"""
if hasattr(body, "__html__"):
body = body.__html__()
elif hasattr(body, "_repr_html_"):
body = body._repr_html_()
return HTTPResponse(
if not isinstance(body, (str, bytes)):
if hasattr(body, "__html__"):
body = body.__html__()
elif hasattr(body, "_repr_html_"):
body = body._repr_html_()
return HTTPResponse( # type: ignore
body,
status=status,
headers=headers,
@ -231,13 +365,13 @@ def html(body, status=200, headers=None):
async def file(
location,
status=200,
mime_type=None,
headers=None,
filename=None,
_range=None,
):
location: Union[str, PurePath],
status: int = 200,
mime_type: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
filename: Optional[str] = None,
_range: Optional[Range] = None,
) -> HTTPResponse:
"""Return a response object with file data.
:param location: Location of file on system.
@ -274,15 +408,15 @@ async def file(
async def file_stream(
location,
status=200,
chunk_size=4096,
mime_type=None,
headers=None,
filename=None,
location: Union[str, PurePath],
status: int = 200,
chunk_size: int = 4096,
mime_type: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
filename: Optional[str] = None,
chunked="deprecated",
_range=None,
):
_range: Optional[Range] = None,
) -> StreamingHTTPResponse:
"""Return a streaming response object with file data.
:param location: Location of file on system.
@ -341,10 +475,10 @@ async def file_stream(
def stream(
streaming_fn,
status=200,
headers=None,
content_type="text/plain; charset=utf-8",
streaming_fn: StreamingFunction,
status: int = 200,
headers: Optional[Dict[str, str]] = None,
content_type: str = "text/plain; charset=utf-8",
chunked="deprecated",
):
"""Accepts an coroutine `streaming_fn` which can be used to
@ -381,15 +515,19 @@ def stream(
def redirect(
to, headers=None, status=302, content_type="text/html; charset=utf-8"
):
"""Abort execution and cause a 302 redirect (by default).
to: str,
headers: Optional[Dict[str, str]] = None,
status: int = 302,
content_type: str = "text/html; charset=utf-8",
) -> HTTPResponse:
"""
Abort execution and cause a 302 redirect (by default) by setting a
Location header.
:param to: path or fully qualified URL to redirect to
:param headers: optional dict of headers to include in the new request
:param status: status code (int) of the new request, defaults to 302
:param content_type: the content type (string) of the response
:returns: the redirecting Response
"""
headers = headers or {}

75
sanic/router.py
View File

@ -1,18 +1,46 @@
from functools import lru_cache
from typing import Any, Dict, Iterable, Optional, Tuple, Union
from sanic_routing import BaseRouter
from sanic_routing.route import Route
from sanic.constants import HTTP_METHODS
from sanic.handlers import RouteHandler
from sanic.request import Request
class Router(BaseRouter):
"""
The router implementation responsible for routing a :class:`Request` object
to the appropriate handler.
"""
DEFAULT_METHOD = "GET"
ALLOWED_METHODS = HTTP_METHODS
@lru_cache
def get(self, request: Request):
def get(
self, request: Request
) -> Tuple[
RouteHandler,
Tuple[Any, ...],
Dict[str, Any],
str,
str,
Optional[str],
bool,
]:
"""
Retrieve a `Route` object containg the details about how to handle
a response for a given request
:param request: the incoming request object
:type request: Request
:return: details needed for handling the request and returning the
correct response
:rtype: Tuple[ RouteHandler, Tuple[Any, ...], Dict[str, Any], str, str,
Optional[str], bool, ]
"""
route, handler, params = self.resolve(
path=request.path,
method=request.method,
@ -34,16 +62,43 @@ class Router(BaseRouter):
def add(
self,
uri,
methods,
handler,
host=None,
strict_slashes=False,
stream=False,
ignore_body=False,
version=None,
name=None,
uri: str,
methods: Iterable[str],
handler: RouteHandler,
host: Optional[str] = None,
strict_slashes: bool = False,
stream: bool = False,
ignore_body: bool = False,
version: Union[str, float, int] = None,
name: Optional[str] = None,
) -> Route:
"""
Add a handler to the router
:param uri: the path of the route
:type uri: str
:param methods: the types of HTTP methods that should be attached,
example: ``["GET", "POST", "OPTIONS"]``
:type methods: Iterable[str]
:param handler: the sync or async function to be executed
:type handler: RouteHandler
:param host: host that the route should be on, defaults to None
:type host: Optional[str], optional
:param strict_slashes: whether to apply strict slashes, defaults
to False
:type strict_slashes: bool, optional
:param stream: whether to stream the response, defaults to False
:type stream: bool, optional
:param ignore_body: whether the incoming request body should be read,
defaults to False
:type ignore_body: bool, optional
:param version: a version modifier for the uri, defaults to None
:type version: Union[str, float, int], optional
:param name: an identifying name of the route, defaults to None
:type name: Optional[str], optional
:return: the route object
:rtype: Route
"""
# TODO: Implement
# - host
# - strict_slashes

13
sanic/server.py
View File

@ -1,19 +1,19 @@
from __future__ import annotations
from typing import (
Optional,
TYPE_CHECKING,
DefaultDict,
Dict,
List,
NamedTuple,
Optional,
Tuple,
Dict,
Type,
Union,
)
if TYPE_CHECKING:
from sanic.http import Http
from sanic.app import Sanic
import asyncio
@ -24,21 +24,22 @@ import socket
import stat
import sys
from sanic.http import Stage
from asyncio import CancelledError
from asyncio.transports import BaseTransport
from functools import partial
from inspect import isawaitable
from ipaddress import ip_address
from signal import SIG_IGN, SIGINT, SIGTERM, Signals
from signal import signal as signal_func
from time import monotonic as current_time
from sanic.request import Request
from sanic.compat import OS_IS_WINDOWS, ctrlc_workaround_for_windows
from sanic.config import Config
from sanic.exceptions import RequestTimeout, ServiceUnavailable
from sanic.http import Http, Stage
from sanic.log import logger
from asyncio.transports import BaseTransport
from sanic.request import Request
try:
import uvloop # type: ignore