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

Conversion of User Guide to the SHH stack (#2781)

Browse Source
This commit is contained in:
Adam Hopkins
2023-09-06 15:44:00 +03:00
committed by GitHub
parent 47215d4635
commit d255d1aae1
332 changed files with 51495 additions and 2013 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
sanic/mixins/base.py
View File

@@ -4,6 +4,8 @@ from sanic.base.meta import SanicMeta
class BaseMixin(metaclass=SanicMeta):
"""Base class for some other mixins."""
name: str
strict_slashes: Optional[bool]

94
sanic/mixins/exceptions.py
View File

@@ -1,4 +1,4 @@
from typing import Set
from typing import Any, Callable, List, Set, Type, Union
from sanic.base.meta import SanicMeta
from sanic.models.futures import FutureException
@@ -11,18 +11,58 @@ class ExceptionMixin(metaclass=SanicMeta):
def _apply_exception_handler(self, handler: FutureException):
raise NotImplementedError # noqa
def exception(self, *exceptions, apply=True):
"""
This method enables the process of creating a global exception
handler for the current blueprint under question.
def exception(
self,
*exceptions: Union[Type[Exception], List[Type[Exception]]],
apply: bool = True
) -> Callable:
"""Decorator used to register an exception handler for the current application or blueprint instance.
:param args: List of Python exceptions to be caught by the handler
:param kwargs: Additional optional arguments to be passed to the
exception handler
This method allows you to define a handler for specific exceptions that
may be raised within the routes of this blueprint. You can specify one
or more exception types to catch, and the handler will be applied to
those exceptions.
:return a decorated method to handle global exceptions for any
route registered under this blueprint.
"""
When used on a Blueprint, the handler will only be applied to routes
registered under that blueprint. That means they only apply to
requests that have been matched, and the exception is raised within
the handler function (or middleware) for that route.
A general exception like `NotFound` should only be registered on the
application instance, not on a blueprint.
See [Exceptions](/en/guide/best-practices/exceptions.html) for more information.
Args:
exceptions (Union[Type[Exception], List[Type[Exception]]]): List of
Python exceptions to be caught by the handler.
apply (bool, optional): Whether the exception handler should be
applied. Defaults to True.
Returns:
Callable: A decorated method to handle global exceptions for any route
registered under this blueprint.
Example:
```python
from sanic import Blueprint, text
bp = Blueprint('my_blueprint')
@bp.exception(Exception)
def handle_exception(request, exception):
return text("Oops, something went wrong!", status=500)
```
```python
from sanic import Sanic, NotFound, text
app = Sanic('MyApp')
@app.exception(NotFound)
def ignore_404s(request, exception):
return text(f"Yep, I totally found the page: {request.url}")
""" # noqa: E501
def decorator(handler):
nonlocal apply
@@ -39,14 +79,30 @@ class ExceptionMixin(metaclass=SanicMeta):
return decorator
def all_exceptions(self, handler):
"""
This method enables the process of creating a global exception
handler for the current blueprint under question.
def all_exceptions(
self, handler: Callable[..., Any]
) -> Callable[..., Any]:
"""Enables the process of creating a global exception handler as a convenience.
:param handler: A coroutine function to handle exceptions
This following two examples are equivalent:
:return a decorated method to handle global exceptions for any
route registered under this blueprint.
"""
```python
@app.exception(Exception)
async def handler(request: Request, exception: Exception) -> HTTPResponse:
return text(f"Exception raised: {exception}")
```
```python
@app.all_exceptions
async def handler(request: Request, exception: Exception) -> HTTPResponse:
return text(f"Exception raised: {exception}")
```
Args:
handler (Callable[..., Any]): A coroutine function to handle exceptions.
Returns:
Callable[..., Any]: A decorated method to handle global exceptions for
any route registered under this blueprint.
""" # noqa: E501
return self.exception(Exception)(handler)

272
sanic/mixins/listeners.py
View File

@@ -59,26 +59,57 @@ class ListenerMixin(metaclass=SanicMeta):
ListenerType[Sanic],
Callable[[ListenerType[Sanic]], ListenerType[Sanic]],
]:
"""
Create a listener from a decorated function.
"""Create a listener for a specific event in the application's lifecycle.
To be used as a decorator:
See [Listeners](/en/guide/basics/listeners) for more details.
.. code-block:: python
.. note::
Overloaded signatures allow for different ways of calling this method, depending on the types of the arguments.
Usually, it is prederred to use one of the convenience methods such as `before_server_start` or `after_server_stop` instead of calling this method directly.
```python
@app.before_server_start
async def prefered_method(_):
...
@app.listener("before_server_start")
async def not_prefered_method(_):
...
Args:
listener_or_event (Union[ListenerType[Sanic], str]): A listener function or an event name.
event_or_none (Optional[str]): The event name to listen for if `listener_or_event` is a function. Defaults to `None`.
apply (bool): Whether to apply the listener immediately. Defaults to `True`.
Returns:
Union[ListenerType[Sanic], Callable[[ListenerType[Sanic]], ListenerType[Sanic]]]: The listener or a callable that takes a listener.
Example:
The following code snippet shows how you can use this method as a decorator:
```python
@bp.listener("before_server_start")
async def before_server_start(app, loop):
...
`See user guide re: listeners
<https://sanicframework.org/guide/basics/listeners.html#listeners>`__
:param event: event to listen to
"""
```
""" # noqa: E501
def register_listener(
listener: ListenerType[Sanic], event: str
) -> ListenerType[Sanic]:
"""A helper function to register a listener for an event.
Typically will not be called directly.
Args:
listener (ListenerType[Sanic]): The listener function to
register.
event (str): The event name to listen for.
Returns:
ListenerType[Sanic]: The listener function that was registered.
"""
nonlocal apply
future_listener = FutureListener(listener, event)
@@ -99,54 +130,275 @@ class ListenerMixin(metaclass=SanicMeta):
def main_process_start(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the main_process_start event.
This event is fired only on the main process and **NOT** on any
worker processes. You should typically use this event to initialize
resources that are shared across workers, or to initialize resources
that are not safe to be initialized in a worker process.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.main_process_start
async def on_main_process_start(app: Sanic):
print("Main process started")
```
""" # noqa: E501
return self.listener(listener, "main_process_start")
def main_process_ready(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the main_process_ready event.
This event is fired only on the main process and **NOT** on any
worker processes. It is fired after the main process has started and
the Worker Manager has been initialized (ie, you will have access to
`app.manager` instance). The typical use case for this event is to
add a managed process to the Worker Manager.
See [Running custom processes](/en/guide/deployment/manager.html#running-custom-processes) and [Listeners](/en/guide/basics/listeners.html) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.main_process_ready
async def on_main_process_ready(app: Sanic):
print("Main process ready")
```
""" # noqa: E501
return self.listener(listener, "main_process_ready")
def main_process_stop(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the main_process_stop event.
This event is fired only on the main process and **NOT** on any
worker processes. You should typically use this event to clean up
resources that were initialized in the main_process_start event.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.main_process_stop
async def on_main_process_stop(app: Sanic):
print("Main process stopped")
```
""" # noqa: E501
return self.listener(listener, "main_process_stop")
def reload_process_start(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the reload_process_start event.
This event is fired only on the reload process and **NOT** on any
worker processes. This is similar to the main_process_start event,
except that it is fired only when the reload process is started.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.reload_process_start
async def on_reload_process_start(app: Sanic):
print("Reload process started")
```
""" # noqa: E501
return self.listener(listener, "reload_process_start")
def reload_process_stop(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the reload_process_stop event.
This event is fired only on the reload process and **NOT** on any
worker processes. This is similar to the main_process_stop event,
except that it is fired only when the reload process is stopped.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.reload_process_stop
async def on_reload_process_stop(app: Sanic):
print("Reload process stopped")
```
""" # noqa: E501
return self.listener(listener, "reload_process_stop")
def before_reload_trigger(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the before_reload_trigger event.
This event is fired only on the reload process and **NOT** on any
worker processes. This event is fired before the reload process
triggers the reload. A change event has been detected and the reload
process is about to be triggered.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.before_reload_trigger
async def on_before_reload_trigger(app: Sanic):
print("Before reload trigger")
```
""" # noqa: E501
return self.listener(listener, "before_reload_trigger")
def after_reload_trigger(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the after_reload_trigger event.
This event is fired only on the reload process and **NOT** on any
worker processes. This event is fired after the reload process
triggers the reload. A change event has been detected and the reload
process has been triggered.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.after_reload_trigger
async def on_after_reload_trigger(app: Sanic, changed: set[str]):
print("After reload trigger, changed files: ", changed)
```
""" # noqa: E501
return self.listener(listener, "after_reload_trigger")
def before_server_start(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the before_server_start event.
This event is fired on all worker processes. You should typically
use this event to initialize resources that are global in nature, or
will be shared across requests and various parts of the application.
A common use case for this event is to initialize a database connection
pool, or to initialize a cache client.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.before_server_start
async def on_before_server_start(app: Sanic):
print("Before server start")
```
""" # noqa: E501
return self.listener(listener, "before_server_start")
def after_server_start(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the after_server_start event.
This event is fired on all worker processes. You should typically
use this event to run background tasks, or perform other actions that
are not directly related to handling requests. In theory, it is
possible that some requests may be handled before this event is fired,
so you should not use this event to initialize resources that are
required for handling requests.
A common use case for this event is to start a background task that
periodically performs some action, such as clearing a cache or
performing a health check.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.after_server_start
async def on_after_server_start(app: Sanic):
print("After server start")
```
""" # noqa: E501
return self.listener(listener, "after_server_start")
def before_server_stop(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the before_server_stop event.
This event is fired on all worker processes. This event is fired
before the server starts shutting down. You should not use this event
to perform any actions that are required for handling requests, as
some requests may continue to be handled after this event is fired.
A common use case for this event is to stop a background task that
was started in the after_server_start event.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.before_server_stop
async def on_before_server_stop(app: Sanic):
print("Before server stop")
```
""" # noqa: E501
return self.listener(listener, "before_server_stop")
def after_server_stop(
self, listener: ListenerType[Sanic]
) -> ListenerType[Sanic]:
"""Decorator for registering a listener for the after_server_stop event.
This event is fired on all worker processes. This event is fired
after the server has stopped shutting down, and all requests have
been handled. You should typically use this event to clean up
resources that were initialized in the before_server_start event.
A common use case for this event is to close a database connection
pool, or to close a cache client.
See [Listeners](/en/guide/basics/listeners) for more details.
Args:
listener (ListenerType[Sanic]): The listener handler to attach.
Examples:
```python
@app.after_server_stop
async def on_after_server_stop(app: Sanic):
print("After server stop")
```
""" # noqa: E501
return self.listener(listener, "after_server_stop")

144
sanic/mixins/middleware.py
View File

@@ -1,11 +1,11 @@
from collections import deque
from functools import partial
from operator import attrgetter
from typing import List
from typing import Callable, List, Union, overload
from sanic.base.meta import SanicMeta
from sanic.middleware import Middleware, MiddlewareLocation
from sanic.models.futures import FutureMiddleware
from sanic.models.futures import FutureMiddleware, MiddlewareType
from sanic.router import Router
@@ -18,24 +18,68 @@ class MiddlewareMixin(metaclass=SanicMeta):
def _apply_middleware(self, middleware: FutureMiddleware):
raise NotImplementedError # noqa
@overload
def middleware(
self,
middleware_or_request,
attach_to="request",
apply=True,
middleware_or_request: MiddlewareType,
attach_to: str = "request",
apply: bool = True,
*,
priority=0
):
"""
Decorate and register middleware to be called before a request
is handled or after a response is created. Can either be called as
*@app.middleware* or *@app.middleware('request')*.
priority: int = 0
) -> MiddlewareType:
...
`See user guide re: middleware
<https://sanicframework.org/guide/basics/middleware.html>`__
@overload
def middleware(
self,
middleware_or_request: str,
attach_to: str = "request",
apply: bool = True,
*,
priority: int = 0
) -> Callable[[MiddlewareType], MiddlewareType]:
...
:param: middleware_or_request: Optional parameter to use for
identifying which type of middleware is being registered.
def middleware(
self,
middleware_or_request: Union[MiddlewareType, str],
attach_to: str = "request",
apply: bool = True,
*,
priority: int = 0
) -> Union[MiddlewareType, Callable[[MiddlewareType], MiddlewareType]]:
"""Decorator for registering middleware.
Decorate and register middleware to be called before a request is
handled or after a response is created. Can either be called as
*@app.middleware* or *@app.middleware('request')*. Although, it is
recommended to use *@app.on_request* or *@app.on_response* instead
for clarity and convenience.
See [Middleware](/guide/basics/middleware) for more information.
Args:
middleware_or_request (Union[Callable, str]): Middleware function
or the keyword 'request' or 'response'.
attach_to (str, optional): When to apply the middleware;
either 'request' (before the request is handled) or 'response'
(after the response is created). Defaults to `'request'`.
apply (bool, optional): Whether the middleware should be applied.
Defaults to `True`.
priority (int, optional): The priority level of the middleware.
Lower numbers are executed first. Defaults to `0`.
Returns:
Union[Callable, Callable[[Callable], Callable]]: The decorated
middleware function or a partial function depending on how
the method was called.
Example:
```python
@app.middleware('request')
async def custom_middleware(request):
...
```
"""
def register_middleware(middleware, attach_to="request"):
@@ -63,17 +107,30 @@ class MiddlewareMixin(metaclass=SanicMeta):
register_middleware, attach_to=middleware_or_request
)
def on_request(self, middleware=None, *, priority=0):
def on_request(self, middleware=None, *, priority=0) -> MiddlewareType:
"""Register a middleware to be called before a request is handled.
This is the same as *@app.middleware('request')*.
:param: middleware: A callable that takes in request.
Args:
middleware (Callable, optional): A callable that takes in a
request. Defaults to `None`.
Returns:
Callable: The decorated middleware function or a partial function
depending on how the method was called.
Examples:
```python
@app.on_request
async def custom_middleware(request):
request.ctx.custom = 'value'
```
"""
if callable(middleware):
return self.middleware(middleware, "request", priority=priority)
else:
return partial(
return partial( # type: ignore
self.middleware, attach_to="request", priority=priority
)
@@ -82,8 +139,20 @@ class MiddlewareMixin(metaclass=SanicMeta):
This is the same as *@app.middleware('response')*.
:param: middleware:
A callable that takes in a request and its response.
Args:
middleware (Callable, optional): A callable that takes in a
request and response. Defaults to `None`.
Returns:
Callable: The decorated middleware function or a partial function
depending on how the method was called.
Examples:
```python
@app.on_response
async def custom_middleware(request, response):
response.headers['X-Server'] = 'Sanic'
```
"""
if callable(middleware):
return self.middleware(middleware, "response", priority=priority)
@@ -92,16 +161,37 @@ class MiddlewareMixin(metaclass=SanicMeta):
self.middleware, attach_to="response", priority=priority
)
def finalize_middleware(self):
def finalize_middleware(self) -> None:
"""Finalize the middleware configuration for the Sanic application.
This method completes the middleware setup for the application.
Middleware in Sanic is used to process requests globally before they
reach individual routes or after routes have been processed.
Finalization consists of identifying defined routes and optimizing
Sanic's performance to meet the application's specific needs. If
you are manually adding routes, after Sanic has started, you will
typically want to use the `amend` context manager rather than
calling this method directly.
.. note::
This method is usually called internally during the server setup
process and does not typically need to be invoked manually.
Example:
```python
app.finalize_middleware()
```
"""
for route in self.router.routes:
request_middleware = Middleware.convert(
self.request_middleware,
self.named_request_middleware.get(route.name, deque()),
self.request_middleware, # type: ignore
self.named_request_middleware.get(route.name, deque()), # type: ignore # noqa: E501
location=MiddlewareLocation.REQUEST,
)
response_middleware = Middleware.convert(
self.response_middleware,
self.named_response_middleware.get(route.name, deque()),
self.response_middleware, # type: ignore
self.named_response_middleware.get(route.name, deque()), # type: ignore # noqa: E501
location=MiddlewareLocation.RESPONSE,
)
route.extra.request_middleware = deque(
@@ -119,11 +209,11 @@ class MiddlewareMixin(metaclass=SanicMeta):
)[::-1]
)
request_middleware = Middleware.convert(
self.request_middleware,
self.request_middleware, # type: ignore
location=MiddlewareLocation.REQUEST,
)
response_middleware = Middleware.convert(
self.response_middleware,
self.response_middleware, # type: ignore
location=MiddlewareLocation.RESPONSE,
)
self.request_middleware = deque(

488
sanic/mixins/routes.py
View File

@@ -58,32 +58,52 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
) -> RouteWrapper:
"""
Decorate a function to be registered as a route
"""Decorate a function to be registered as a route.
Args:
uri (str): Path of the URL.
methods (Optional[Iterable[str]]): List or tuple of
methods allowed.
host (Optional[Union[str, List[str]]]): The host, if required.
strict_slashes (Optional[bool]): Whether to apply strict slashes
to the route.
stream (bool): Whether to allow the request to stream its body.
version (Optional[Union[int, str, float]]): Route specific
versioning.
name (Optional[str]): User-defined route name for url_for.
ignore_body (bool): Whether the handler should ignore request
body (e.g. `GET` requests).
apply (bool): Apply middleware to the route.
subprotocols (Optional[List[str]]): List of subprotocols.
websocket (bool): Enable WebSocket support.
unquote (bool): Unquote special characters in the URL path.
static (bool): Enable static route.
version_prefix (str): URL path that should be before the version
value; default: `"/v"`.
error_format (Optional[str]): Error format for the route.
ctx_kwargs (Any): Keyword arguments that begin with a `ctx_*`
prefix will be appended to the route context (`route.ctx`).
**Example using context kwargs**
Returns:
RouteWrapper: Tuple of routes, decorated function.
.. code-block:: python
Examples:
Using the method to define a GET endpoint:
@app.route(..., ctx_foo="foobar")
async def route_handler(request: Request):
assert request.route.ctx.foo == "foobar"
```python
@app.route("/hello")
async def hello(request: Request):
return text("Hello, World!")
```
:param uri: path of the URL
:param methods: list or tuple of methods allowed
:param host: the host, if required
:param strict_slashes: whether to apply strict slashes to the route
:param stream: whether to allow the request to stream its body
:param version: route specific versioning
:param name: user defined route name for url_for
:param ignore_body: whether the handler should ignore request
body (eg. GET requests)
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: tuple of routes, decorated function
Adding context kwargs to the route:
```python
@app.route("/greet", ctx_name="World")
async def greet(request: Request):
name = request.route.ctx.name
return text(f"Hello, {name}!")
```
"""
# Fix case where the user did not prefix the URL with a /
@@ -209,25 +229,56 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
unquote: bool = False,
**ctx_kwargs: Any,
) -> RouteHandler:
"""A helper method to register class instance or
functions as a handler to the application url
routes.
"""A helper method to register class-based view or functions as a handler to the application url routes.
:param handler: function or class instance
:param uri: path of the URL
:param methods: list or tuple of methods allowed, these are overridden
if using a HTTPMethodView
:param host:
:param strict_slashes:
:param version:
:param name: user defined route name for url_for
:param stream: boolean specifying if the handler is a stream handler
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: function or class instance
"""
Args:
handler (RouteHandler): Function or class-based view used as a route handler.
uri (str): Path of the URL.
methods (Iterable[str]): List or tuple of methods allowed; these are overridden if using an HTTPMethodView.
host (Optional[Union[str, List[str]]]): Hostname or hostnames to match for this route.
strict_slashes (Optional[bool]): If set, a route's slashes will be strict. E.g. `/foo` will not match `/foo/`.
version (Optional[Union[int, str, float]]): Version of the API for this route.
name (Optional[str]): User-defined route name for `url_for`.
stream (bool): Boolean specifying if the handler is a stream handler.
version_prefix (str): URL path that should be before the version value; default: ``/v``.
error_format (Optional[str]): Custom error format string.
unquote (bool): Boolean specifying if the handler requires unquoting.
ctx_kwargs (Any): Keyword arguments that begin with a `ctx_*` prefix will be appended to the route context (``route.ctx``). See below for examples.
Returns:
RouteHandler: The route handler.
Examples:
```python
from sanic import Sanic, text
app = Sanic("test")
async def handler(request):
return text("OK")
app.add_route(handler, "/test", methods=["GET", "POST"])
```
You can use `ctx_kwargs` to add custom context to the route. This
can often be useful when wanting to add metadata to a route that
can be used by other parts of the application (like middleware).
```python
from sanic import Sanic, text
app = Sanic("test")
async def handler(request):
return text("OK")
async def custom_middleware(request):
if request.route.ctx.monitor:
do_some_monitoring()
app.add_route(handler, "/test", methods=["GET", "POST"], ctx_monitor=True)
app.register_middleware(custom_middleware)
""" # noqa: E501
# Handle HTTPMethodView differently
if hasattr(handler, "view_class"):
methods = set()
@@ -271,21 +322,31 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
) -> RouteHandler:
"""
Add an API URL under the **GET** *HTTP* method
"""Decorate a function handler to create a route definition using the **GET** HTTP method.
:param uri: URL to be tagged to **GET** method of *HTTP*
:param host: Host IP or FQDN for the service to use
:param strict_slashes: Instruct :class:`Sanic` to check if the request
URLs need to terminate with a */*
:param version: API Version
:param name: Unique name that can be used to identify the Route
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
Args:
uri (str): URL to be tagged to GET method of HTTP.
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
the service to use.
strict_slashes (Optional[bool]): Instruct Sanic to check if the
request URLs need to terminate with a `/`.
version (Optional[Union[int, str, float]]): API Version.
name (Optional[str]): Unique name that can be used to identify
the route.
ignore_body (bool): Whether the handler should ignore request
body. This means the body of the request, if sent, will not
be consumed. In that instance, you will see a warning in
the logs. Defaults to `True`, meaning do not consume the body.
version_prefix (str): URL path that should be before the version
value. Defaults to `"/v"`.
error_format (Optional[str]): Custom error format string.
**ctx_kwargs (Any): Keyword arguments that begin with a
`ctx_* prefix` will be appended to the route
context (`route.ctx`).
Returns:
RouteHandler: Object decorated with route method.
""" # noqa: E501
return cast(
RouteHandler,
self.route(
@@ -314,21 +375,29 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
) -> RouteHandler:
"""
Add an API URL under the **POST** *HTTP* method
"""Decorate a function handler to create a route definition using the **POST** HTTP method.
:param uri: URL to be tagged to **POST** method of *HTTP*
:param host: Host IP or FQDN for the service to use
:param strict_slashes: Instruct :class:`Sanic` to check if the request
URLs need to terminate with a */*
:param version: API Version
:param name: Unique name that can be used to identify the Route
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
Args:
uri (str): URL to be tagged to POST method of HTTP.
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
the service to use.
strict_slashes (Optional[bool]): Instruct Sanic to check if the
request URLs need to terminate with a `/`.
stream (bool): Whether or not to stream the request body.
Defaults to `False`.
version (Optional[Union[int, str, float]]): API Version.
name (Optional[str]): Unique name that can be used to identify
the route.
version_prefix (str): URL path that should be before the version
value. Defaults to `"/v"`.
error_format (Optional[str]): Custom error format string.
**ctx_kwargs (Any): Keyword arguments that begin with a
`ctx_*` prefix will be appended to the route
context (`route.ctx`).
Returns:
RouteHandler: Object decorated with route method.
""" # noqa: E501
return cast(
RouteHandler,
self.route(
@@ -357,21 +426,29 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
) -> RouteHandler:
"""
Add an API URL under the **PUT** *HTTP* method
"""Decorate a function handler to create a route definition using the **PUT** HTTP method.
:param uri: URL to be tagged to **PUT** method of *HTTP*
:param host: Host IP or FQDN for the service to use
:param strict_slashes: Instruct :class:`Sanic` to check if the request
URLs need to terminate with a */*
:param version: API Version
:param name: Unique name that can be used to identify the Route
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
Args:
uri (str): URL to be tagged to PUT method of HTTP.
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
the service to use.
strict_slashes (Optional[bool]): Instruct Sanic to check if the
request URLs need to terminate with a `/`.
stream (bool): Whether or not to stream the request body.
Defaults to `False`.
version (Optional[Union[int, str, float]]): API Version.
name (Optional[str]): Unique name that can be used to identify
the route.
version_prefix (str): URL path that should be before the version
value. Defaults to `"/v"`.
error_format (Optional[str]): Custom error format string.
**ctx_kwargs (Any): Keyword arguments that begin with a
`ctx_*` prefix will be appended to the route
context (`route.ctx`).
Returns:
RouteHandler: Object decorated with route method.
""" # noqa: E501
return cast(
RouteHandler,
self.route(
@@ -400,29 +477,31 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
) -> RouteHandler:
"""
Add an API URL under the **HEAD** *HTTP* method
"""Decorate a function handler to create a route definition using the **HEAD** HTTP method.
:param uri: URL to be tagged to **HEAD** method of *HTTP*
:type uri: str
:param host: Host IP or FQDN for the service to use
:type host: Optional[str], optional
:param strict_slashes: Instruct :class:`Sanic` to check if the request
URLs need to terminate with a */*
:type strict_slashes: Optional[bool], optional
:param version: API Version
:type version: Optional[str], optional
:param name: Unique name that can be used to identify the Route
:type name: Optional[str], optional
:param ignore_body: whether the handler should ignore request
body (eg. GET requests), defaults to True
:type ignore_body: bool, optional
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
Args:
uri (str): URL to be tagged to HEAD method of HTTP.
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
the service to use.
strict_slashes (Optional[bool]): Instruct Sanic to check if the
request URLs need to terminate with a `/`.
version (Optional[Union[int, str, float]]): API Version.
name (Optional[str]): Unique name that can be used to identify
the route.
ignore_body (bool): Whether the handler should ignore request
body. This means the body of the request, if sent, will not
be consumed. In that instance, you will see a warning in
the logs. Defaults to `True`, meaning do not consume the body.
version_prefix (str): URL path that should be before the version
value. Defaults to `"/v"`.
error_format (Optional[str]): Custom error format string.
**ctx_kwargs (Any): Keyword arguments that begin with a
`ctx_*` prefix will be appended to the route
context (`route.ctx`).
Returns:
RouteHandler: Object decorated with route method.
""" # noqa: E501
return cast(
RouteHandler,
self.route(
@@ -451,29 +530,31 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
) -> RouteHandler:
"""
Add an API URL under the **OPTIONS** *HTTP* method
"""Decorate a function handler to create a route definition using the **OPTIONS** HTTP method.
:param uri: URL to be tagged to **OPTIONS** method of *HTTP*
:type uri: str
:param host: Host IP or FQDN for the service to use
:type host: Optional[str], optional
:param strict_slashes: Instruct :class:`Sanic` to check if the request
URLs need to terminate with a */*
:type strict_slashes: Optional[bool], optional
:param version: API Version
:type version: Optional[str], optional
:param name: Unique name that can be used to identify the Route
:type name: Optional[str], optional
:param ignore_body: whether the handler should ignore request
body (eg. GET requests), defaults to True
:type ignore_body: bool, optional
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
Args:
uri (str): URL to be tagged to OPTIONS method of HTTP.
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
the service to use.
strict_slashes (Optional[bool]): Instruct Sanic to check if the
request URLs need to terminate with a `/`.
version (Optional[Union[int, str, float]]): API Version.
name (Optional[str]): Unique name that can be used to identify
the route.
ignore_body (bool): Whether the handler should ignore request
body. This means the body of the request, if sent, will not
be consumed. In that instance, you will see a warning in
the logs. Defaults to `True`, meaning do not consume the body.
version_prefix (str): URL path that should be before the version
value. Defaults to `"/v"`.
error_format (Optional[str]): Custom error format string.
**ctx_kwargs (Any): Keyword arguments that begin with a
`ctx_*` prefix will be appended to the route
context (`route.ctx`).
Returns:
RouteHandler: Object decorated with route method.
""" # noqa: E501
return cast(
RouteHandler,
self.route(
@@ -502,31 +583,29 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
) -> RouteHandler:
"""
Add an API URL under the **PATCH** *HTTP* method
"""Decorate a function handler to create a route definition using the **PATCH** HTTP method.
:param uri: URL to be tagged to **PATCH** method of *HTTP*
:type uri: str
:param host: Host IP or FQDN for the service to use
:type host: Optional[str], optional
:param strict_slashes: Instruct :class:`Sanic` to check if the request
URLs need to terminate with a */*
:type strict_slashes: Optional[bool], optional
:param stream: whether to allow the request to stream its body
:type stream: Optional[bool], optional
:param version: API Version
:type version: Optional[str], optional
:param name: Unique name that can be used to identify the Route
:type name: Optional[str], optional
:param ignore_body: whether the handler should ignore request
body (eg. GET requests), defaults to True
:type ignore_body: bool, optional
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
Args:
uri (str): URL to be tagged to PATCH method of HTTP.
host (Optional[Union[str, List[str]]]): Host IP or FQDN for
the service to use.
strict_slashes (Optional[bool]): Instruct Sanic to check if the
request URLs need to terminate with a `/`.
stream (bool): Set to `True` if full request streaming is needed,
`False` otherwise. Defaults to `False`.
version (Optional[Union[int, str, float]]): API Version.
name (Optional[str]): Unique name that can be used to identify
the route.
version_prefix (str): URL path that should be before the version
value. Defaults to `"/v"`.
error_format (Optional[str]): Custom error format string.
**ctx_kwargs (Any): Keyword arguments that begin with a
`ctx_*` prefix will be appended to the route
context (`route.ctx`).
Returns:
RouteHandler: Object decorated with route method.
""" # noqa: E501
return cast(
RouteHandler,
self.route(
@@ -555,21 +634,28 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
) -> RouteHandler:
"""
Add an API URL under the **DELETE** *HTTP* method
"""Decorate a function handler to create a route definition using the **DELETE** HTTP method.
:param uri: URL to be tagged to **DELETE** method of *HTTP*
:param host: Host IP or FQDN for the service to use
:param strict_slashes: Instruct :class:`Sanic` to check if the request
URLs need to terminate with a */*
:param version: API Version
:param name: Unique name that can be used to identify the Route
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
Args:
uri (str): URL to be tagged to the DELETE method of HTTP.
host (Optional[Union[str, List[str]]]): Host IP or FQDN for the
service to use.
strict_slashes (Optional[bool]): Instruct Sanic to check if the
request URLs need to terminate with a */*.
version (Optional[Union[int, str, float]]): API Version.
name (Optional[str]): Unique name that can be used to identify
the Route.
ignore_body (bool): Whether or not to ignore the body in the
request. Defaults to `False`.
version_prefix (str): URL path that should be before the version
value. Defaults to `"/v"`.
error_format (Optional[str]): Custom error format string.
**ctx_kwargs (Any): Keyword arguments that begin with a `ctx_*`
prefix will be appended to the route context (`route.ctx`).
Returns:
RouteHandler: Object decorated with route method.
""" # noqa: E501
return cast(
RouteHandler,
self.route(
@@ -599,21 +685,30 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
):
"""
Decorate a function to be registered as a websocket route
"""Decorate a function to be registered as a websocket route.
:param uri: path of the URL
:param host: Host IP or FQDN details
:param strict_slashes: If the API endpoint needs to terminate
with a "/" or not
:param subprotocols: optional list of str with supported subprotocols
:param name: A unique name assigned to the URL so that it can
be used with :func:`url_for`
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: tuple of routes, decorated function
Args:
uri (str): Path of the URL.
host (Optional[Union[str, List[str]]]): Host IP or FQDN details.
strict_slashes (Optional[bool]): If the API endpoint needs to
terminate with a `"/"` or not.
subprotocols (Optional[List[str]]): Optional list of str with
supported subprotocols.
version (Optional[Union[int, str, float]]): WebSocket
protocol version.
name (Optional[str]): A unique name assigned to the URL so that
it can be used with url_for.
apply (bool): If set to False, it doesn't apply the route to the
app. Default is `True`.
version_prefix (str): URL path that should be before the version
value. Defaults to `"/v"`.
error_format (Optional[str]): Custom error format string.
**ctx_kwargs (Any): Keyword arguments that begin with
a `ctx_* prefix` will be appended to the route
context (`route.ctx`).
Returns:
tuple: Tuple of routes, decorated function.
"""
return self.route(
uri=uri,
@@ -643,26 +738,27 @@ class RouteMixin(BaseMixin, metaclass=SanicMeta):
error_format: Optional[str] = None,
**ctx_kwargs: Any,
):
"""
A helper method to register a function as a websocket route.
"""A helper method to register a function as a websocket route.
:param handler: a callable function or instance of a class
that can handle the websocket request
:param host: Host IP or FQDN details
:param uri: URL path that will be mapped to the websocket
handler
handler
:param strict_slashes: If the API endpoint needs to terminate
with a "/" or not
:param subprotocols: Subprotocols to be used with websocket
handshake
:param name: A unique name assigned to the URL so that it can
be used with :func:`url_for`
:param version_prefix: URL path that should be before the version
value; default: ``/v``
:param ctx_kwargs: Keyword arguments that begin with a ctx_* prefix
will be appended to the route context (``route.ctx``)
:return: Objected decorated by :func:`websocket`
Args:
handler (Callable): A callable function or instance of a class
that can handle the websocket request.
uri (str): URL path that will be mapped to the websocket handler.
host (Optional[Union[str, List[str]]]): Host IP or FQDN details.
strict_slashes (Optional[bool]): If the API endpoint needs to
terminate with a `"/"` or not.
subprotocols (Optional[List[str]]): Subprotocols to be used with
websocket handshake.
version (Optional[Union[int, str, float]]): Versioning information.
name (Optional[str]): A unique name assigned to the URL.
version_prefix (str): URL path before the version value.
Defaults to `"/v"`.
error_format (Optional[str]): Format for error handling.
**ctx_kwargs (Any): Keyword arguments beginning with `ctx_*`
prefix will be appended to the route context (`route.ctx`).
Returns:
Callable: Object passed as the handler.
"""
return self.websocket(
uri=uri,

59
sanic/mixins/signals.py
View File

@@ -1,5 +1,7 @@
from __future__ import annotations
from enum import Enum
from typing import Any, Callable, Dict, Optional, Set, Union
from typing import Any, Callable, Coroutine, Dict, Optional, Set, Union
from sanic.base.meta import SanicMeta
from sanic.models.futures import FutureSignal
@@ -66,7 +68,24 @@ class SignalMixin(metaclass=SanicMeta):
event: str,
condition: Optional[Dict[str, Any]] = None,
exclusive: bool = True,
):
) -> Callable[..., Any]:
"""Registers a signal handler for a specific event.
Args:
handler (Optional[Callable[..., Any]]): The function to be called
when the event occurs. Defaults to a noop if not provided.
event (str): The name of the event to listen for.
condition (Optional[Dict[str, Any]]): Optional condition to filter
the event triggering. Defaults to `None`.
exclusive (bool): Whether or not the handler is exclusive. When
`True`, the signal can only be dispatched when the
`condition` has been met. *This is inapplicable to blueprint
signals, which are **ALWAYS** non-exclusive.* Defaults
to `True`.
Returns:
Callable[..., Any]: The handler that was registered.
"""
if not handler:
async def noop():
@@ -81,8 +100,40 @@ class SignalMixin(metaclass=SanicMeta):
def event(self, event: str):
raise NotImplementedError
def catch_exception(self, handler):
def catch_exception(
self,
handler: Callable[[SignalMixin, Exception], Coroutine[Any, Any, None]],
) -> None:
"""Register an exception handler for logging or processing.
This method allows the registration of a custom exception handler to
catch and process exceptions that occur in the application. Unlike a
typical exception handler that might modify the response to the client,
this is intended to capture exceptions for logging or other internal
processing, such as sending them to an error reporting utility.
Args:
handler (Callable): A coroutine function that takes the application
instance and the exception as arguments. It will be called when
an exception occurs within the application's lifecycle.
Example:
```python
app = Sanic("TestApp")
@app.catch_exception
async def report_exception(app: Sanic, exception: Exception):
logging.error(f"An exception occurred: {exception}")
# Send to an error reporting service
await error_service.report(exception)
# Any unhandled exceptions within the application will now be
# logged and reported to the error service.
```
""" # noqa: E501
async def signal_handler(exception: Exception):
await handler(self, exception)
self.signal(Event.SERVER_LIFECYCLE_EXCEPTION)(signal_handler)
self.signal(Event.SERVER_EXCEPTION_REPORT)(signal_handler)

439
sanic/mixins/startup.py
View File

@@ -54,7 +54,7 @@ from sanic.helpers import Default, _default, is_atty
from sanic.http.constants import HTTP
from sanic.http.tls import get_ssl_context, process_to_context
from sanic.http.tls.context import SanicSSLContext
from sanic.log import Colors, error_logger, logger
from sanic.log import Colors, deprecation, error_logger, logger
from sanic.models.handler_types import ListenerType
from sanic.server import Signal as ServerSignal
from sanic.server import try_use_uvloop
@@ -90,6 +90,7 @@ else: # no cov
class StartupMixin(metaclass=SanicMeta):
_app_registry: ClassVar[Dict[str, Sanic]]
asgi: bool
config: Config
listeners: Dict[str, List[ListenerType[Any]]]
state: ApplicationState
@@ -100,7 +101,15 @@ class StartupMixin(metaclass=SanicMeta):
start_method: ClassVar[StartMethod] = _default
START_METHOD_SET: ClassVar[bool] = False
def setup_loop(self):
def setup_loop(self) -> None:
"""Set up the event loop.
An internal method that sets up the event loop to uvloop if
possible, or a Windows selector loop if on Windows.
Returns:
None
"""
if not self.asgi:
if self.config.USE_UVLOOP is True or (
isinstance(self.config.USE_UVLOOP, Default)
@@ -112,10 +121,39 @@ class StartupMixin(metaclass=SanicMeta):
@property
def m(self) -> WorkerMultiplexer:
"""Interface for interacting with the worker processes"""
"""Interface for interacting with the worker processes
This is a shortcut for `app.multiplexer`. It is available only in a
worker process using the Sanic server. It allows you to interact with
the worker processes, such as sending messages and commands.
See [Access to the multiplexer](/en/guide/deployment/manager#access-to-the-multiplexer) for more information.
Returns:
WorkerMultiplexer: The worker multiplexer instance
Examples:
```python
app.m.restart() # restarts the worker
app.m.terminate() # terminates the worker
app.m.scale(4) # scales the number of workers to 4
```
""" # noqa: E501
return self.multiplexer
def make_coffee(self, *args, **kwargs):
"""
Try for yourself! `sanic server:app --coffee`
```
▄████████▄
██ ██▀▀▄
███████████ █
███████████▄▄▀
▀███████▀
```
"""
self.state.coffee = True
self.run(*args, **kwargs)
@@ -146,42 +184,77 @@ class StartupMixin(metaclass=SanicMeta):
auto_tls: bool = False,
single_process: bool = False,
) -> None:
"""
Run the HTTP Server and listen until keyboard interrupt or term
signal. On termination, drain connections before closing.
"""Run the HTTP Server and listen until keyboard interrupt or term signal. On termination, drain connections before closing.
:param host: Address to host on
:type host: str
:param port: Port to host on
:type port: int
:param debug: Enables debug output (slows server)
:type debug: bool
:param auto_reload: Reload app whenever its source code is changed.
Enabled by default in debug mode.
:type auto_relaod: bool
:param ssl: SSLContext, or location of certificate and key
for SSL encryption of worker(s)
:type ssl: str, dict, SSLContext or list
:param sock: Socket for the server to accept connections from
:type sock: socket
:param workers: Number of processes received before it is respected
:type workers: int
:param protocol: Subclass of asyncio Protocol class
:type protocol: type[Protocol]
:param backlog: a number of unaccepted connections that the system
will allow before refusing new connections
:type backlog: int
:param register_sys_signals: Register SIG* events
:type register_sys_signals: bool
:param access_log: Enables writing access logs (slows server)
:type access_log: bool
:param unix: Unix socket to listen on instead of TCP port
:type unix: str
:param noisy_exceptions: Log exceptions that are normally considered
to be quiet/silent
:type noisy_exceptions: bool
:return: Nothing
"""
.. note::
When you need control over running the Sanic instance, this is the method to use.
However, in most cases the preferred method is to use the CLI command:
```sh
sanic server:app`
```
If you are using this method to run Sanic, make sure you do the following:
1. Use `if __name__ == "__main__"` to guard the code.
2. Do **NOT** define the app instance inside the `if` block.
See [Dynamic Applications](/en/guide/deployment/app-loader) for more information about the second point.
Args:
host (Optional[str]): Address to host on.
port (Optional[int]): Port to host on.
dev (bool): Run the server in development mode.
debug (bool): Enables debug output (slows server).
auto_reload (Optional[bool]): Reload app whenever its source code is changed.
Enabled by default in debug mode.
version (HTTPVersion): HTTP Version.
ssl (Union[None, SSLContext, dict, str, list, tuple]): SSLContext, or location of certificate and key
for SSL encryption of worker(s).
sock (Optional[socket]): Socket for the server to accept connections from.
workers (int): Number of processes received before it is respected.
protocol (Optional[Type[Protocol]]): Subclass of asyncio Protocol class.
backlog (int): A number of unaccepted connections that the system will allow
before refusing new connections.
register_sys_signals (bool): Register SIG* events.
access_log (Optional[bool]): Enables writing access logs (slows server).
unix (Optional[str]): Unix socket to listen on instead of TCP port.
loop (Optional[AbstractEventLoop]): AsyncIO event loop.
reload_dir (Optional[Union[List[str], str]]): Directory to watch for code changes, if auto_reload is True.
noisy_exceptions (Optional[bool]): Log exceptions that are normally considered to be quiet/silent.
motd (bool): Display Message of the Day.
fast (bool): Enable fast mode.
verbosity (int): Verbosity level.
motd_display (Optional[Dict[str, str]]): Customize Message of the Day display.
auto_tls (bool): Enable automatic TLS certificate handling.
single_process (bool): Enable single process mode.
Returns:
None
Raises:
RuntimeError: Raised when attempting to serve HTTP/3 as a secondary server.
RuntimeError: Raised when attempting to use both `fast` and `workers`.
RuntimeError: Raised when attempting to use `single_process` with `fast`, `workers`, or `auto_reload`.
TypeError: Raised when attempting to use `loop` with `create_server`.
ValueError: Raised when `PROXIES_COUNT` is negative.
Examples:
```python
from sanic import Sanic, Request, json
app = Sanic("TestApp")
@app.get("/")
async def handler(request: Request):
return json({"foo": "bar"})
if __name__ == "__main__":
app.run(port=9999, dev=True)
```
""" # noqa: E501
self.prepare(
host=host,
port=port,
@@ -242,6 +315,53 @@ class StartupMixin(metaclass=SanicMeta):
auto_tls: bool = False,
single_process: bool = False,
) -> None:
"""Prepares one or more Sanic applications to be served simultaneously.
This low-level API is typically used when you need to run multiple Sanic applications at the same time. Once prepared, `Sanic.serve()` should be called in the `if __name__ == "__main__"` block.
.. note::
"Preparing" and "serving" with this function is equivalent to using `app.run` for a single instance. This should only be used when running multiple applications at the same time.
Args:
host (Optional[str], optional): Hostname to listen on. Defaults to `None`.
port (Optional[int], optional): Port to listen on. Defaults to `None`.
dev (bool, optional): Development mode. Defaults to `False`.
debug (bool, optional): Debug mode. Defaults to `False`.
auto_reload (Optional[bool], optional): Auto reload feature. Defaults to `None`.
version (HTTPVersion, optional): HTTP version to use. Defaults to `HTTP.VERSION_1`.
ssl (Union[None, SSLContext, dict, str, list, tuple], optional): SSL configuration. Defaults to `None`.
sock (Optional[socket], optional): Socket to bind to. Defaults to `None`.
workers (int, optional): Number of worker processes. Defaults to `1`.
protocol (Optional[Type[Protocol]], optional): Custom protocol class. Defaults to `None`.
backlog (int, optional): Maximum number of pending connections. Defaults to `100`.
register_sys_signals (bool, optional): Register system signals. Defaults to `True`.
access_log (Optional[bool], optional): Access log. Defaults to `None`.
unix (Optional[str], optional): Unix socket. Defaults to `None`.
loop (Optional[AbstractEventLoop], optional): Event loop. Defaults to `None`.
reload_dir (Optional[Union[List[str], str]], optional): Reload directory. Defaults to `None`.
noisy_exceptions (Optional[bool], optional): Display exceptions. Defaults to `None`.
motd (bool, optional): Display message of the day. Defaults to `True`.
fast (bool, optional): Fast mode. Defaults to `False`.
verbosity (int, optional): Verbosity level. Defaults to `0`.
motd_display (Optional[Dict[str, str]], optional): Custom MOTD display. Defaults to `None`.
coffee (bool, optional): Coffee mode. Defaults to `False`.
auto_tls (bool, optional): Auto TLS. Defaults to `False`.
single_process (bool, optional): Single process mode. Defaults to `False`.
Raises:
RuntimeError: Raised when attempting to serve HTTP/3 as a secondary server.
RuntimeError: Raised when attempting to use both `fast` and `workers`.
RuntimeError: Raised when attempting to use `single_process` with `fast`, `workers`, or `auto_reload`.
TypeError: Raised when attempting to use `loop` with `create_server`.
ValueError: Raised when `PROXIES_COUNT` is negative.
Examples:
```python
if __name__ == "__main__":
app.prepare()
app.serve()
```
""" # noqa: E501
if version == 3 and self.state.server_info:
raise RuntimeError(
"Serving HTTP/3 instances as a secondary server is "
@@ -361,50 +481,77 @@ class StartupMixin(metaclass=SanicMeta):
backlog: int = 100,
access_log: Optional[bool] = None,
unix: Optional[str] = None,
return_asyncio_server: bool = False,
return_asyncio_server: bool = True,
asyncio_server_kwargs: Optional[Dict[str, Any]] = None,
noisy_exceptions: Optional[bool] = None,
) -> Optional[AsyncioServer]:
"""
Asynchronous version of :func:`run`.
Low level API for creating a Sanic Server instance.
This method will take care of the operations necessary to invoke
the *before_start* events via :func:`trigger_events` method invocation
before starting the *sanic* app in Async mode.
This method will create a Sanic Server instance, but will not start
it. This is useful for integrating Sanic into other systems. But, you
should take caution when using it as it is a low level API and does
not perform any of the lifecycle events.
.. note::
This does not support multiprocessing and is not the preferred
way to run a :class:`Sanic` application.
way to run a Sanic application. Proceed with caution.
:param host: Address to host on
:type host: str
:param port: Port to host on
:type port: int
:param debug: Enables debug output (slows server)
:type debug: bool
:param ssl: SSLContext, or location of certificate and key
for SSL encryption of worker(s)
:type ssl: SSLContext or dict
:param sock: Socket for the server to accept connections from
:type sock: socket
:param protocol: Subclass of asyncio Protocol class
:type protocol: type[Protocol]
:param backlog: a number of unaccepted connections that the system
will allow before refusing new connections
:type backlog: int
:param access_log: Enables writing access logs (slows server)
:type access_log: bool
:param return_asyncio_server: flag that defines whether there's a need
to return asyncio.Server or
start it serving right away
:type return_asyncio_server: bool
:param asyncio_server_kwargs: key-value arguments for
asyncio/uvloop create_server method
:type asyncio_server_kwargs: dict
:param noisy_exceptions: Log exceptions that are normally considered
to be quiet/silent
:type noisy_exceptions: bool
:return: AsyncioServer if return_asyncio_server is true, else Nothing
You will need to start the server yourself as shown in the example
below. You are responsible for the lifecycle of the server, including
app startup using `await app.startup()`. No events will be triggered
for you, so you will need to trigger them yourself if wanted.
Args:
host (Optional[str]): Address to host on.
port (Optional[int]): Port to host on.
debug (bool): Enables debug output (slows server).
ssl (Union[None, SSLContext, dict, str, list, tuple]): SSLContext,
or location of certificate and key for SSL encryption
of worker(s).
sock (Optional[socket]): Socket for the server to accept
connections from.
protocol (Optional[Type[Protocol]]): Subclass of
`asyncio.Protocol` class.
backlog (int): Number of unaccepted connections that the system
will allow before refusing new connections.
access_log (Optional[bool]): Enables writing access logs
(slows server).
return_asyncio_server (bool): _DEPRECATED_
asyncio_server_kwargs (Optional[Dict[str, Any]]): Key-value
arguments for asyncio/uvloop `create_server` method.
noisy_exceptions (Optional[bool]): Log exceptions that are normally
considered to be quiet/silent.
Returns:
Optional[AsyncioServer]: AsyncioServer if `return_asyncio_server`
is `True` else `None`.
Examples:
```python
import asyncio
import uvloop
from sanic import Sanic, response
app = Sanic("Example")
@app.route("/")
async def test(request):
return response.json({"answer": "42"})
async def main():
server = await app.create_server()
await server.startup()
await server.serve_forever()
if __name__ == "__main__":
asyncio.set_event_loop(uvloop.new_event_loop())
asyncio.run(main())
```
"""
if sock is None:
@@ -423,6 +570,14 @@ class StartupMixin(metaclass=SanicMeta):
if value is not None:
setattr(self.config, attribute, value)
if not return_asyncio_server:
return_asyncio_server = True
deprecation(
"The `return_asyncio_server` argument is deprecated and "
"ignored. It will be removed in v24.3.",
24.3,
)
server_settings = self._helper(
host=host,
port=port,
@@ -456,9 +611,16 @@ class StartupMixin(metaclass=SanicMeta):
asyncio_server_kwargs=asyncio_server_kwargs, **server_settings
)
def stop(self, terminate: bool = True, unregister: bool = False):
"""
This kills the Sanic
def stop(self, terminate: bool = True, unregister: bool = False) -> None:
"""This kills the Sanic server, cleaning up after itself.
Args:
terminate (bool): Force kill all requests immediately without
allowing them to finish processing.
unregister (bool): Unregister the app from the global registry.
Returns:
None
"""
if terminate and hasattr(self, "multiplexer"):
self.multiplexer.terminate()
@@ -565,7 +727,19 @@ class StartupMixin(metaclass=SanicMeta):
def motd(
self,
server_settings: Optional[Dict[str, Any]] = None,
):
) -> None:
"""Outputs the message of the day (MOTD).
It generally can only be called once per process, and is usually
called by the `run` method in the main process.
Args:
server_settings (Optional[Dict[str, Any]], optional): Settings for
the server. Defaults to `None`.
Returns:
None
"""
if (
os.environ.get("SANIC_WORKER_NAME")
or os.environ.get("SANIC_MOTD_OUTPUT")
@@ -583,6 +757,17 @@ class StartupMixin(metaclass=SanicMeta):
def get_motd_data(
self, server_settings: Optional[Dict[str, Any]] = None
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""Retrieves the message of the day (MOTD) data.
Args:
server_settings (Optional[Dict[str, Any]], optional): Settings for
the server. Defaults to `None`.
Returns:
Tuple[Dict[str, Any], Dict[str, Any]]: A tuple containing two
dictionaries with the relevant MOTD data.
"""
mode = [f"{self.state.mode},"]
if self.state.fast:
mode.append("goin' fast")
@@ -646,6 +831,11 @@ class StartupMixin(metaclass=SanicMeta):
@property
def serve_location(self) -> str:
"""Retrieve the server location.
Returns:
str: The server location.
"""
try:
server_settings = self.state.server_info[0].settings
return self.get_server_location(server_settings)
@@ -657,6 +847,15 @@ class StartupMixin(metaclass=SanicMeta):
def get_server_location(
server_settings: Optional[Dict[str, Any]] = None
) -> str:
"""Using the server settings, retrieve the server location.
Args:
server_settings (Optional[Dict[str, Any]], optional): Settings for
the server. Defaults to `None`.
Returns:
str: The server location.
"""
serve_location = ""
proto = "http"
if not server_settings:
@@ -686,12 +885,29 @@ class StartupMixin(metaclass=SanicMeta):
version: HTTPVersion = HTTP.VERSION_1,
auto_tls: bool = False,
) -> Tuple[str, int]:
"""Retrieve the host address and port, with default values based on the given parameters.
Args:
host (Optional[str]): Host IP or FQDN for the service to use. Defaults to `"127.0.0.1"`.
port (Optional[int]): Port number. Defaults to `8443` if version is 3 or `auto_tls=True`, else `8000`
version (HTTPVersion, optional): HTTP Version. Defaults to `HTTP.VERSION_1` (HTTP/1.1).
auto_tls (bool, optional): Automatic TLS flag. Defaults to `False`.
Returns:
Tuple[str, int]: Tuple containing the host and port
""" # noqa: E501
host = host or "127.0.0.1"
port = port or (8443 if (version == 3 or auto_tls) else 8000)
return host, port
@classmethod
def should_auto_reload(cls) -> bool:
"""Check if any applications have auto-reload enabled.
Returns:
bool: `True` if any applications have auto-reload enabled, else
`False`.
"""
return any(app.state.auto_reload for app in cls._app_registry.values())
@classmethod
@@ -731,6 +947,42 @@ class StartupMixin(metaclass=SanicMeta):
app_loader: Optional[AppLoader] = None,
factory: Optional[Callable[[], Sanic]] = None,
) -> None:
"""Serve one or more Sanic applications.
This is the main entry point for running Sanic applications. It
should be called in the `if __name__ == "__main__"` block.
Args:
primary (Optional[Sanic], optional): The primary Sanic application
to serve. Defaults to `None`.
app_loader (Optional[AppLoader], optional): An AppLoader instance
to use for loading applications. Defaults to `None`.
factory (Optional[Callable[[], Sanic]], optional): A factory
function to use for loading applications. Defaults to `None`.
Raises:
RuntimeError: Raised when no applications are found.
RuntimeError: Raised when no server information is found for the
primary application.
RuntimeError: Raised when attempting to use `loop` with
`create_server`.
RuntimeError: Raised when attempting to use `single_process` with
`fast`, `workers`, or `auto_reload`.
RuntimeError: Raised when attempting to serve HTTP/3 as a
secondary server.
RuntimeError: Raised when attempting to use both `fast` and
`workers`.
TypeError: Raised when attempting to use `loop` with
`create_server`.
ValueError: Raised when `PROXIES_COUNT` is negative.
Examples:
```python
if __name__ == "__main__":
app.prepare()
Sanic.serve()
```
"""
cls._set_startup_method()
os.environ["SANIC_MOTD_OUTPUT"] = "true"
apps = list(cls._app_registry.values())
@@ -913,6 +1165,39 @@ class StartupMixin(metaclass=SanicMeta):
@classmethod
def serve_single(cls, primary: Optional[Sanic] = None) -> None:
"""Serve a single process of a Sanic application.
Similar to `serve`, but only serves a single process. When used,
certain features are disabled, such as `fast`, `workers`,
`multiplexer`, `auto_reload`, and the Inspector. It is almost
never needed to use this method directly. Instead, you should
use the CLI:
```sh
sanic app.sanic:app --single-process
```
Or, if you need to do it programmatically, you should use the
`single_process` argument of `run`:
```python
app.run(single_process=True)
```
Args:
primary (Optional[Sanic], optional): The primary Sanic application
to serve. Defaults to `None`.
Raises:
RuntimeError: Raised when no applications are found.
RuntimeError: Raised when no server information is found for the
primary application.
RuntimeError: Raised when attempting to serve HTTP/3 as a
secondary server.
RuntimeError: Raised when attempting to use both `fast` and
`workers`.
ValueError: Raised when `PROXIES_COUNT` is negative.
"""
os.environ["SANIC_MOTD_OUTPUT"] = "true"
apps = list(cls._app_registry.values())

100
sanic/mixins/static.py
View File

@@ -46,43 +46,71 @@ class StaticMixin(BaseMixin, metaclass=SanicMeta):
directory_view: bool = False,
directory_handler: Optional[DirectoryHandler] = None,
):
"""
Register a root to serve files from. The input can either be a
file or a directory. This method will enable an easy and simple way
to setup the :class:`Route` necessary to serve the static files.
"""Register a root to serve files from. The input can either be a file or a directory.
:param uri: URL path to be used for serving static content
:param file_or_directory: Path for the Static file/directory with
static files
:param pattern: Regex Pattern identifying the valid static files
:param use_modified_since: If true, send file modified time, and return
not modified if the browser's matches the server's
:param use_content_range: If true, process header for range requests
and sends the file part that is requested
:param stream_large_files: If true, use the
:func:`StreamingHTTPResponse.file_stream` handler rather
than the :func:`HTTPResponse.file` handler to send the file.
If this is an integer, this represents the threshold size to
switch to :func:`StreamingHTTPResponse.file_stream`
:param name: user defined name used for url_for
:param host: Host IP or FQDN for the service to use
:param strict_slashes: Instruct :class:`Sanic` to check if the request
URLs need to terminate with a */*
:param content_type: user defined content type for header
:param apply: If true, will register the route immediately
:param resource_type: Explicitly declare a resource to be a "
file" or a "dir"
:param index: When exposing against a directory, index is the name that
will be served as the default file. When multiple files names are
passed, then they will be tried in order.
:param directory_view: Whether to fallback to showing the directory
viewer when exposing a directory
:param directory_handler: An instance of :class:`DirectoryHandler`
that can be used for explicitly controlling and subclassing the
behavior of the default directory handler
:return: routes registered on the router
:rtype: List[sanic.router.Route]
"""
This method provides an easy and simple way to set up the route necessary to serve static files.
Args:
uri (str): URL path to be used for serving static content.
file_or_directory (Union[PathLike, str]): Path to the static file
or directory with static files.
pattern (str, optional): Regex pattern identifying the valid
static files. Defaults to `r"/?.+"`.
use_modified_since (bool, optional): If true, send file modified
time, and return not modified if the browser's matches the
server's. Defaults to `True`.
use_content_range (bool, optional): If true, process header for
range requests and sends the file part that is requested.
Defaults to `False`.
stream_large_files (Union[bool, int], optional): If `True`, use
the `StreamingHTTPResponse.file_stream` handler rather than
the `HTTPResponse.file handler` to send the file. If this
is an integer, it represents the threshold size to switch
to `StreamingHTTPResponse.file_stream`. Defaults to `False`,
which means that the response will not be streamed.
name (str, optional): User-defined name used for url_for.
Defaults to `"static"`.
host (Optional[str], optional): Host IP or FQDN for the
service to use.
strict_slashes (Optional[bool], optional): Instruct Sanic to
check if the request URLs need to terminate with a slash.
content_type (Optional[str], optional): User-defined content type
for header.
apply (bool, optional): If true, will register the route
immediately. Defaults to `True`.
resource_type (Optional[str], optional): Explicitly declare a
resource to be a `"file"` or a `"dir"`.
index (Optional[Union[str, Sequence[str]]], optional): When
exposing against a directory, index is the name that will
be served as the default file. When multiple file names are
passed, then they will be tried in order.
directory_view (bool, optional): Whether to fallback to showing
the directory viewer when exposing a directory. Defaults
to `False`.
directory_handler (Optional[DirectoryHandler], optional): An
instance of DirectoryHandler that can be used for explicitly
controlling and subclassing the behavior of the default
directory handler.
Returns:
List[sanic.router.Route]: Routes registered on the router.
Examples:
Serving a single file:
```python
app.static('/foo', 'path/to/static/file.txt')
```
Serving all files from a directory:
```python
app.static('/static', 'path/to/static/directory')
```
Serving large files with a specific threshold:
```python
app.static('/static', 'path/to/large/files', stream_large_files=1000000)
```
""" # noqa: E501
name = self._generate_name(name)