# Copyright 2019-2021 by Kurt Griffiths
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""ASGI application class."""
from __future__ import annotations
import asyncio
from inspect import isasyncgenfunction
from inspect import iscoroutinefunction
import traceback
from typing import (
Any,
Awaitable,
Callable,
ClassVar,
Dict,
Iterable,
List,
Optional,
overload,
Tuple,
Type,
TYPE_CHECKING,
TypeVar,
Union,
)
from falcon import constants
from falcon import responders
from falcon import routing
from falcon._typing import _UNSET
from falcon._typing import AsgiErrorHandler
from falcon._typing import AsgiReceive
from falcon._typing import AsgiResponderCallable
from falcon._typing import AsgiResponderWsCallable
from falcon._typing import AsgiSend
from falcon._typing import AsgiSinkCallable
from falcon._typing import SinkPrefix
import falcon.app
from falcon.app_helpers import AsyncPreparedMiddlewareResult
from falcon.app_helpers import AsyncPreparedMiddlewareWsResult
from falcon.app_helpers import prepare_middleware
from falcon.app_helpers import prepare_middleware_ws
from falcon.asgi_spec import AsgiSendMsg
from falcon.asgi_spec import EventType
from falcon.asgi_spec import WSCloseCode
from falcon.constants import MEDIA_JSON
from falcon.errors import CompatibilityError
from falcon.errors import HTTPBadRequest
from falcon.errors import WebSocketDisconnected
from falcon.http_error import HTTPError
from falcon.http_status import HTTPStatus
from falcon.media.multipart import MultipartFormHandler
from falcon.util import get_argnames
from falcon.util.misc import is_python_func
from falcon.util.sync import _should_wrap_non_coroutines
from falcon.util.sync import _wrap_non_coroutine_unsafe
from falcon.util.sync import wrap_sync_to_async
from ._asgi_helpers import _validate_asgi_scope
from ._asgi_helpers import _wrap_asgi_coroutine_func
from .multipart import MultipartForm
from .request import Request
from .response import Response
from .structures import SSEvent
from .ws import _supports_reason
from .ws import http_status_to_ws_code
from .ws import WebSocket
from .ws import WebSocketOptions
__all__ = ('App',)
# TODO(vytas): Clean up these foul workarounds before the 4.0 release.
MultipartFormHandler._ASGI_MULTIPART_FORM = MultipartForm
_EVT_RESP_EOF: AsgiSendMsg = {'type': EventType.HTTP_RESPONSE_BODY}
_BODILESS_STATUS_CODES = frozenset([100, 101, 204, 304])
_TYPELESS_STATUS_CODES = frozenset([204, 304])
_FALLBACK_WS_ERROR_CODE = 3011
_BE = TypeVar('_BE', bound=BaseException)
class App(falcon.app.App):
'''The main entry point into a Falcon-based ASGI app.
Each App instance provides a callable
`ASGI `_ interface
and a routing engine (for WSGI applications, see
:class:`falcon.App`).
Keyword Arguments:
media_type (str): Default media type to use when initializing
:class:`~.RequestOptions` and
:class:`~.ResponseOptions`. The ``falcon``
module provides a number of constants for common media types,
such as ``falcon.MEDIA_MSGPACK``, ``falcon.MEDIA_YAML``,
``falcon.MEDIA_XML``, etc.
middleware: Either a single middleware component object or an iterable
of objects (instantiated classes) that implement the
middleware component interface shown below.
The interface provides support for handling both ASGI worker
lifespan events and per-request events. However, because lifespan
events are an optional part of the ASGI specification, they may or
may not fire depending on your ASGI server.
A lifespan event handler can be used to perform startup or shutdown
activities for the main event loop. An example of this would be
creating a connection pool and subsequently closing the connection
pool to release the connections.
Note:
In a multi-process environment, lifespan events will be
triggered independently for the individual event loop associated
with each process.
Note:
The framework requires that all middleware methods be
implemented as coroutine functions via `async def`. However,
it is possible to implement middleware classes that support
both ASGI and WSGI apps by distinguishing the ASGI methods
with an `*_async` postfix (see also:
:ref:`Middleware `).
It is only necessary to implement the methods for the events you
would like to handle; Falcon simply skips over any missing
middleware methods::
class ExampleMiddleware:
async def process_startup(
self, scope: dict[str, Any], event: dict[str, Any]
) -> None:
"""Process the ASGI lifespan startup event.
Invoked when the server is ready to start up and
receive connections, but before it has started to
do so.
To halt startup processing and signal to the server that it
should terminate, simply raise an exception and the
framework will convert it to a "lifespan.startup.failed"
event for the server.
Args:
scope (dict): The ASGI scope dictionary for the
lifespan protocol. The lifespan scope exists
for the duration of the event loop.
event (dict): The ASGI event dictionary for the
startup event.
"""
async def process_shutdown(
self, scope: dict[str, Any], event: dict[str, Any]
) -> None:
"""Process the ASGI lifespan shutdown event.
Invoked when the server has stopped accepting
connections and closed all active connections.
To halt shutdown processing and signal to the server
that it should immediately terminate, simply raise an
exception and the framework will convert it to a
"lifespan.shutdown.failed" event for the server.
Args:
scope (dict): The ASGI scope dictionary for the
lifespan protocol. The lifespan scope exists
for the duration of the event loop.
event (dict): The ASGI event dictionary for the
shutdown event.
"""
async def process_request(
self, req: Request, resp: Response
) -> None:
"""Process the request before routing it.
Note:
Because Falcon routes each request based on
req.path, a request can be effectively re-routed
by setting that attribute to a new value from
within process_request().
Args:
req: Request object that will eventually be
routed to an on_* responder method.
resp: Response object that will be routed to
the on_* responder.
"""
async def process_resource(
self,
req: Request,
resp: Response,
resource: object,
params: dict[str, Any],
) -> None:
"""Process the request and resource *after* routing.
Note:
This method is only called when the request matches
a route to a resource.
Args:
req: Request object that will be passed to the
routed responder.
resp: Response object that will be passed to the
responder.
resource: Resource object to which the request was
routed. May be ``None`` if no route was found for
the request.
params: A dict-like object representing any
additional params derived from the route's URI
template fields, that will be passed to the
resource's responder method as keyword
arguments.
"""
async def process_response(
self,
req: Request,
resp: Response,
resource: object,
req_succeeded: bool
) -> None:
"""Post-processing of the response (after routing).
Args:
req: Request object.
resp: Response object.
resource: Resource object to which the request was
routed. May be ``None`` if no route was found
for the request.
req_succeeded: True if no exceptions were raised
while the framework processed and routed the
request; otherwise False.
"""
# WebSocket methods
async def process_request_ws(
self, req: Request, ws: WebSocket
) -> None:
"""Process a WebSocket handshake request before routing it.
Note:
Because Falcon routes each request based on req.path, a
request can be effectively re-routed by setting that
attribute to a new value from within process_request().
Args:
req: Request object that will eventually be
passed into an on_websocket() responder method.
ws: The WebSocket object that will be passed into
on_websocket() after routing.
"""
async def process_resource_ws(
self,
req: Request,
ws: WebSocket,
resource: object,
params: dict[str, Any],
) -> None:
"""Process a WebSocket handshake request after routing.
Note:
This method is only called when the request matches
a route to a resource.
Args:
req: Request object that will be passed to the
routed responder.
ws: WebSocket object that will be passed to the
routed responder.
resource: Resource object to which the request was
routed.
params: A dict-like object representing any additional
params derived from the route's URI template fields,
that will be passed to the resource's responder
method as keyword arguments.
"""
(See also: :ref:`Middleware `)
request_type: ``Request``-like class to use instead
of Falcon's default class. Among other things, this feature
affords inheriting from :class:`falcon.asgi.Request` in order
to override the ``context_type`` class variable
(default: :class:`falcon.asgi.Request`)
response_type: ``Response``-like class to use
instead of Falcon's default class (default:
:class:`falcon.asgi.Response`)
router (object): An instance of a custom router
to use in lieu of the default engine.
(See also: :ref:`Custom Routers `)
independent_middleware (bool): Set to ``False`` if response
middleware should not be executed independently of whether or
not request middleware raises an exception (default
``True``). When this option is set to ``False``, a middleware
component's ``process_response()`` method will NOT be called
when that same component's ``process_request()`` (or that of
a component higher up in the stack) raises an exception.
cors_enable (bool): Set this flag to ``True`` to enable a simple
CORS policy for all responses, including support for preflighted
requests. An instance of :class:`..CORSMiddleware` can instead be
passed to the middleware argument to customize its behaviour.
(default ``False``).
(See also: :ref:`CORS `)
sink_before_static_route (bool): Indicates if the sinks should be processed
before (when ``True``) or after (when ``False``) the static routes.
This has an effect only if no route was matched. (default ``True``)
'''
_STATIC_ROUTE_TYPE = routing.StaticRouteAsync
# NOTE(kgriffs): This makes it easier to tell what we are dealing with
# without having to import falcon.asgi.
_ASGI: ClassVar[bool] = True
_default_responder_bad_request: ClassVar[AsgiResponderCallable] = (
responders.bad_request_async # type: ignore[assignment]
)
_default_responder_path_not_found: ClassVar[AsgiResponderCallable] = (
responders.path_not_found_async # type: ignore[assignment]
)
__slots__ = (
'_standard_response_type',
'_middleware_ws',
'ws_options',
)
_error_handlers: Dict[Type[BaseException], AsgiErrorHandler] # type: ignore[assignment]
_middleware: AsyncPreparedMiddlewareResult # type: ignore[assignment]
_middleware_ws: AsyncPreparedMiddlewareWsResult
_request_type: Type[Request]
_response_type: Type[Response]
ws_options: WebSocketOptions
"""A set of behavioral options related to WebSocket connections.
See also: :class:`~.WebSocketOptions`.
"""
def __init__(
self,
media_type: str = constants.DEFAULT_MEDIA_TYPE,
request_type: Optional[Type[Request]] = None,
response_type: Optional[Type[Response]] = None,
middleware: Union[object, Iterable[object]] = None,
router: Optional[routing.CompiledRouter] = None,
independent_middleware: bool = True,
cors_enable: bool = False,
sink_before_static_route: bool = True,
) -> None:
super().__init__(
media_type,
request_type or Request,
response_type or Response,
middleware,
router,
independent_middleware,
cors_enable,
sink_before_static_route,
)
self.ws_options = WebSocketOptions()
self._standard_response_type = response_type in (None, Response)
self.add_error_handler(
WebSocketDisconnected, self._ws_disconnected_error_handler
)
@_wrap_asgi_coroutine_func
async def __call__( # type: ignore[override] # noqa: C901
self,
scope: Dict[str, Any],
receive: AsgiReceive,
send: AsgiSend,
) -> None:
# NOTE(kgriffs): The ASGI spec requires the 'type' key to be present.
scope_type: str = scope['type']
# PERF(kgriffs): This should usually be present, so use a
# try..except
try:
asgi_info: Dict[str, str] = scope['asgi']
except KeyError:
# NOTE(kgriffs): According to the ASGI spec, "2.0" is
# the default version.
asgi_info = scope['asgi'] = {'version': '2.0'}
try:
spec_version: Optional[str] = asgi_info['spec_version']
except KeyError:
spec_version = None
try:
http_version: str = scope['http_version']
except KeyError:
http_version = '1.1'
spec_version = _validate_asgi_scope(scope_type, spec_version, http_version)
if scope_type != 'http':
# PERF(vytas): Evaluate the potentially recurring WebSocket path
# first (in contrast to one-shot lifespan events).
if scope_type == 'websocket':
await self._handle_websocket(spec_version, scope, receive, send)
return
# NOTE(vytas): Else 'lifespan' -- other scope_type values have been
# eliminated by _validate_asgi_scope at this point.
await self._call_lifespan_handlers(spec_version, scope, receive, send)
return
# NOTE(kgriffs): Per the ASGI spec, we should not proceed with request
# processing until after we receive an initial 'http.request' event.
first_event = await receive()
first_event_type = first_event['type']
# PERF(vytas): Inline the value of EventType.HTTP_DISCONNECT in this
# critical code path.
if first_event_type == 'http.disconnect':
# NOTE(kgriffs): Bail out immediately to minimize resource usage
return
# NOTE(kgriffs): This is the only other type defined by the ASGI spec,
# but we just assert it to make it easier to track down a potential
# incompatibility with a future spec version.
# PERF(vytas): Inline the value of EventType.HTTP_REQUEST in this
# critical code path.
assert first_event_type == 'http.request'
req = self._request_type(
scope, receive, first_event=first_event, options=self.req_options
)
resp = self._response_type(options=self.resp_options)
resource: Optional[object] = None
params: Dict[str, Any] = {}
dependent_mw_resp_stack: list = []
mw_req_stack, mw_rsrc_stack, mw_resp_stack = self._middleware
req_succeeded = False
try:
if req.method in self._META_METHODS:
raise HTTPBadRequest()
# NOTE(ealogar): The execution of request middleware
# should be before routing. This will allow request mw
# to modify the path.
# NOTE: if flag set to use independent middleware, execute
# request middleware independently. Otherwise, only queue
# response middleware after request middleware succeeds.
if self._independent_middleware:
for process_request in mw_req_stack:
await process_request(req, resp) # type: ignore[operator]
if resp.complete:
break
else:
for process_request, process_response in mw_req_stack: # type: ignore[misc, assignment]
if process_request and not resp.complete:
await process_request(req, resp) # type: ignore[operator]
if process_response:
dependent_mw_resp_stack.insert(0, process_response)
if not resp.complete:
# NOTE(warsaw): Moved this to inside the try except
# because it is possible when using object-based
# traversal for _get_responder() to fail. An example is
# a case where an object does not have the requested
# next-hop child resource. In that case, the object
# being asked to dispatch to its child will raise an
# HTTP exception signaling the problem, e.g. a 404.
responder: AsgiResponderCallable
responder, params, resource, req.uri_template = self._get_responder(req) # type: ignore[assignment]
except Exception as ex:
if not await self._handle_exception(req, resp, ex, params):
raise
else:
try:
# NOTE(kgriffs): If the request did not match any
# route, a default responder is returned and the
# resource is None. In that case, we skip the
# resource middleware methods. Resource will also be
# None when a middleware method already set
# resp.complete to True.
if resource:
# Call process_resource middleware methods.
for process_resource in mw_rsrc_stack:
await process_resource(req, resp, resource, params)
if resp.complete:
break
if not resp.complete:
await responder(req, resp, **params)
req_succeeded = True
except Exception as ex:
if not await self._handle_exception(req, resp, ex, params):
raise
# Call process_response middleware methods.
for process_response in mw_resp_stack or dependent_mw_resp_stack:
try:
await process_response(req, resp, resource, req_succeeded)
except Exception as ex:
if not await self._handle_exception(req, resp, ex, params):
raise
req_succeeded = False
data: Optional[bytes] = b''
try:
# NOTE(vytas): It is only safe to inline Response.render_body()
# where we can be sure it hasn't been overridden, either directly
# or by modifying the behavior of its dependencies.
if self._standard_response_type:
# PERF(vytas): inline Response.render_body() in this critical code
# path in order to shave off an await.
text = resp.text
if text is None:
data = resp._data
if data is None and resp._media is not None:
# NOTE(kgriffs): We use a special _UNSET singleton since
# None is ambiguous (the media handler might return None).
if resp._media_rendered is _UNSET:
opt = resp.options
if not resp.content_type:
resp.content_type = opt.default_media_type
handler, serialize_sync, _ = opt.media_handlers._resolve(
resp.content_type, opt.default_media_type
)
if serialize_sync:
resp._media_rendered = serialize_sync(resp._media)
else:
resp._media_rendered = await handler.serialize_async(
resp._media, resp.content_type
)
data = resp._media_rendered
else:
try:
# NOTE(kgriffs): Normally we expect text to be a string
data = text.encode()
except AttributeError:
# NOTE(kgriffs): Assume it was a bytes object already
data = text # type: ignore[assignment]
else:
# NOTE(vytas): Custom response type.
data = await resp.render_body()
except Exception as ex:
if not await self._handle_exception(req, resp, ex, params):
raise
req_succeeded = False
resp_status: int = resp.status_code
default_media_type: Optional[str] = self.resp_options.default_media_type
if req.method == 'HEAD' or resp_status in _BODILESS_STATUS_CODES:
#
# PERF(vytas): move check for the less common and much faster path
# of resp_status being in {204, 304} here; NB: this builds on the
# assumption _TYPELESS_STATUS_CODES <= _BODILESS_STATUS_CODES.
#
# NOTE(kgriffs): Based on wsgiref.validate's interpretation of
# RFC 2616, as commented in that module's source code. The
# presence of the Content-Length header is not similarly
# enforced.
#
# NOTE(kgriffs): Assuming the same for ASGI until proven otherwise.
#
if resp_status in _TYPELESS_STATUS_CODES:
default_media_type = None
elif (
# NOTE(kgriffs): If they are going to stream using an
# async generator, we can't know in advance what the
# content length will be.
(data is not None or not resp.stream)
and req.method == 'HEAD'
and resp_status not in _BODILESS_STATUS_CODES
and 'content-length' not in resp._headers
):
# NOTE(kgriffs): We really should be returning a Content-Length
# in this case according to my reading of the RFCs. By
# optionally using len(data) we let a resource simulate HEAD
# by turning around and calling it's own on_get().
resp._headers['content-length'] = str(len(data)) if data else '0'
await send(
{
# PERF(vytas): Inline the value of
# EventType.HTTP_RESPONSE_START in this critical code path.
'type': 'http.response.start',
'status': resp_status,
'headers': resp._asgi_headers(default_media_type),
}
)
await send(_EVT_RESP_EOF)
# PERF(vytas): Check resp._registered_callbacks directly to shave
# off a function call since this is a hot/critical code path.
if resp._registered_callbacks:
self._schedule_callbacks(resp)
return
# PERF(vytas): Operate directly on the resp private interface to reduce
# overhead since this is a hot/critical code path.
if resp._sse:
sse_emitter = resp._sse
if isasyncgenfunction(sse_emitter):
raise TypeError(
'Response.sse must be an async iterable. This can be obtained by '
'simply executing the async generator function and then setting '
'the result to Response.sse, e.g.: '
'resp.sse = some_asyncgen_function()'
)
# NOTE(kgriffs): This must be done in a separate task because
# receive() can block for some time (until the connection is
# actually closed).
async def watch_disconnect() -> None:
while True:
received_event = await receive()
if received_event['type'] == EventType.HTTP_DISCONNECT:
break
watcher = asyncio.create_task(watch_disconnect())
await send(
{
'type': EventType.HTTP_RESPONSE_START,
'status': resp_status,
'headers': resp._asgi_headers('text/event-stream'),
}
)
# PERF(vytas): Check resp._registered_callbacks directly to shave
# off a function call since this is a hot/critical code path.
if resp._registered_callbacks:
self._schedule_callbacks(resp)
sse_handler, _, _ = self.resp_options.media_handlers._resolve(
MEDIA_JSON, MEDIA_JSON, raise_not_found=False
)
# TODO(kgriffs): Do we need to do anything special to handle when
# a connection is closed?
async for event in sse_emitter:
if not event:
event = SSEvent()
# NOTE(kgriffs): According to the ASGI spec, once the client
# disconnects, send() acts as a no-op. We have to check
# the connection state using watch_disconnect() above.
await send(
{
'type': EventType.HTTP_RESPONSE_BODY,
'body': event.serialize(sse_handler),
'more_body': True,
}
)
if watcher.done(): # pragma: no py39,py310 cover
break
watcher.cancel()
try:
await watcher
except asyncio.CancelledError:
pass
await send({'type': EventType.HTTP_RESPONSE_BODY})
return
if data is not None:
# PERF(kgriffs): Böse mußt sein. Operate directly on resp._headers
# to reduce overhead since this is a hot/critical code path.
# NOTE(kgriffs): We always set content-length to match the
# body bytes length, even if content-length is already set. The
# reason being that web servers and LBs behave unpredictably
# when the header doesn't match the body (sometimes choosing to
# drop the HTTP connection prematurely, for example).
resp._headers['content-length'] = str(len(data))
await send(
{
# PERF(vytas): Inline the value of
# EventType.HTTP_RESPONSE_START in this critical code path.
'type': 'http.response.start',
'status': resp_status,
'headers': resp._asgi_headers(default_media_type),
}
)
await send(
{
# PERF(vytas): Inline the value of
# EventType.HTTP_RESPONSE_BODY in this critical code path.
'type': 'http.response.body',
'body': data,
}
)
# PERF(vytas): Check resp._registered_callbacks directly to shave
# off a function call since this is a hot/critical code path.
if resp._registered_callbacks:
self._schedule_callbacks(resp)
return
stream = resp.stream
if not stream:
resp._headers['content-length'] = '0'
await send(
{
# PERF(vytas): Inline the value of
# EventType.HTTP_RESPONSE_START in this critical code path.
'type': 'http.response.start',
'status': resp_status,
'headers': resp._asgi_headers(default_media_type),
}
)
if stream:
# Detect whether this is one of the following:
#
# (a) async file-like object (e.g., aiofiles)
# (b) async generator
# (c) async iterator
#
if hasattr(stream, 'read'):
try:
while True:
data = await stream.read(self._STREAM_BLOCK_SIZE)
if data == b'':
break
else:
await send(
{
'type': EventType.HTTP_RESPONSE_BODY,
# NOTE(kgriffs): Handle the case in which
# data is None
'body': data or b'',
'more_body': True,
}
)
finally:
if hasattr(stream, 'close'):
await stream.close()
else:
# NOTE(kgriffs): Works for both async generators and iterators
try:
async for data in stream:
# NOTE(kgriffs): We can not rely on StopIteration
# because of Pep 479 that is implemented starting
# with Python 3.7. AFAICT this is only an issue
# when using an async iterator instead of an async
# generator.
if data is None:
break
await send(
{
'type': EventType.HTTP_RESPONSE_BODY,
'body': data,
'more_body': True,
}
)
except TypeError as ex:
if isasyncgenfunction(stream):
raise TypeError(
'The object assigned to Response.stream appears to '
'be an async generator function. A generator '
'object is expected instead. This can be obtained '
'simply by calling the generator function, e.g.: '
'resp.stream = some_asyncgen_function()'
)
raise TypeError(
'Response.stream must be a generator or implement an '
'__aiter__ method. Error raised while iterating over '
'Response.stream: ' + str(ex)
)
finally:
# NOTE(vytas): This could be DRYed with the above identical
# twoliner in a one large block, but OTOH we would be
# unable to reuse the current try.. except.
if hasattr(stream, 'close'):
await stream.close()
await send(_EVT_RESP_EOF)
# PERF(vytas): Check resp._registered_callbacks directly to shave
# off a function call since this is a hot/critical code path.
if resp._registered_callbacks:
self._schedule_callbacks(resp)
def add_route(self, uri_template: str, resource: object, **kwargs: Any) -> None:
# NOTE(kgriffs): Inject an extra kwarg so that the compiled router
# will know to validate the responder methods to make sure they
# are async coroutines.
kwargs['_asgi'] = True
super().add_route(uri_template, resource, **kwargs)
add_route.__doc__ = falcon.app.App.add_route.__doc__ # NOTE: not really required
def add_sink(self, sink: AsgiSinkCallable, prefix: SinkPrefix = r'/') -> None: # type: ignore[override]
if not iscoroutinefunction(sink) and is_python_func(sink):
if _should_wrap_non_coroutines():
sink = wrap_sync_to_async(sink)
else:
raise CompatibilityError(
'The sink method must be an awaitable coroutine function '
'in order to be used safely with an ASGI app.'
)
super().add_sink(sink, prefix=prefix) # type: ignore[arg-type]
add_sink.__doc__ = falcon.app.App.add_sink.__doc__ # NOTE: not really required
@overload # type: ignore[override]
def add_error_handler(
self,
exception: Type[_BE],
handler: Callable[[Request, Response, _BE, Dict[str, Any]], Awaitable[None]],
) -> None: ...
@overload
def add_error_handler(
self,
exception: Union[Type[BaseException], Iterable[Type[BaseException]]],
handler: Optional[AsgiErrorHandler] = None,
) -> None: ...
def add_error_handler( # type: ignore[misc]
self,
exception: Union[Type[BaseException], Iterable[Type[BaseException]]],
handler: Optional[AsgiErrorHandler] = None,
) -> None:
"""Register a handler for one or more exception types.
Error handlers may be registered for any exception type, including
:class:`~.HTTPError` or :class:`~.HTTPStatus`. This feature
provides a central location for logging and otherwise handling
exceptions raised by responders, hooks, and middleware components.
A handler can raise an instance of :class:`~.HTTPError` or
:class:`~.HTTPStatus` to communicate information about the issue to
the client. Alternatively, a handler may modify `resp`
directly.
An error handler "matches" a raised exception if the exception is an
instance of the corresponding exception type. If more than one error
handler matches the raised exception, the framework will choose the
most specific one, as determined by the method resolution order of the
raised exception type. If multiple error handlers are registered for the
*same* exception class, then the most recently-registered handler is
used.
For example, suppose we register error handlers as follows::
app = App()
app.add_error_handler(falcon.HTTPNotFound, custom_handle_not_found)
app.add_error_handler(falcon.HTTPError, custom_handle_http_error)
app.add_error_handler(Exception, custom_handle_uncaught_exception)
app.add_error_handler(falcon.HTTPNotFound, custom_handle_404)
If an instance of ``falcon.HTTPForbidden`` is raised, it will be
handled by ``custom_handle_http_error()``. ``falcon.HTTPError`` is a
superclass of ``falcon.HTTPForbidden`` and a subclass of ``Exception``,
so it is the most specific exception type with a registered handler.
If an instance of ``falcon.HTTPNotFound`` is raised, it will be handled
by ``custom_handle_404()``, not by ``custom_handle_not_found()``, because
``custom_handle_404()`` was registered more recently.
Note:
By default, the framework installs three handlers, one for
:class:`~.HTTPError`, one for :class:`~.HTTPStatus`, and one for
the standard ``Exception`` type, which prevents passing uncaught
exceptions to the WSGI server. These can be overridden by adding a
custom error handler method for the exception type in question.
When a generic unhandled exception is raised while
handling a :ref:`WebSocket ` connection, the default handler will
close the connection with the standard close code ``1011`` (Internal
Error). If your ASGI server does not support this code, the
framework will use code ``3011`` instead; or you can customize
it via the
:attr:`~falcon.asgi.WebSocketOptions.error_close_code`
property of :attr:`~.ws_options`.
On the other hand, if an ``on_websocket()`` responder raises an
instance of :class:`~falcon.HTTPError`, the default error handler
will close the :ref:`WebSocket ` connection with a framework
close code derived by adding ``3000`` to the HTTP status code (e.g.,
``3404``)
Args:
exception (type or iterable of types): When handling a request,
whenever an error occurs that is an instance of the specified
type(s), the associated handler will be called. Either a single
type or an iterable of types may be specified.
handler (callable): A coroutine function taking the
form::
async def func(req, resp, ex, params, ws=None):
pass
In the case of a WebSocket connection, the `resp` argument
will be ``None``, while the `ws` keyword argument
will receive the :class:`~falcon.asgi.WebSocket` object
representing the connection.
If the `handler` keyword argument is not provided to
:meth:`~.add_error_handler`, the handler will default to
``exception.handle``, where ``exception`` is the error type
specified above, and ``handle`` is a static method (i.e.,
decorated with ``@staticmethod``) that accepts the params
just described. For example::
class CustomException(CustomBaseException):
@staticmethod
async def handle(req, resp, ex, params):
# TODO: Log the error
# Convert to an instance of falcon.HTTPError
raise falcon.HTTPError(falcon.HTTP_792)
Note, however, that if an iterable of exception types is
specified instead of a single type, the handler must be
explicitly specified using the `handler` keyword argument.
"""
if handler is None:
handler = getattr(exception, 'handle', None)
if handler is None:
raise AttributeError(
'handler must either be specified explicitly or defined as a '
'static method named "handle" that is a member of the given '
'exception class.'
)
# NOTE(vytas): Do not shoot ourselves in the foot in case error
# handlers are our own cythonized code.
if handler not in (
self._http_status_handler,
self._http_error_handler,
self._python_error_handler,
):
handler = _wrap_non_coroutine_unsafe(handler) # type: ignore[assignment]
# NOTE(kgriffs): iscoroutinefunction() always returns False
# for cythonized functions.
#
# https://github.com/cython/cython/issues/2273
# https://bugs.python.org/issue38225
#
if not iscoroutinefunction(handler) and is_python_func(handler):
raise CompatibilityError(
'The handler must be an awaitable coroutine function in order '
'to be used safely with an ASGI app.'
)
handler_callable: AsgiErrorHandler = handler
exception_tuple: Tuple[type[BaseException], ...]
try:
exception_tuple = tuple(exception) # type: ignore[arg-type]
except TypeError:
exception_tuple = (exception,) # type: ignore[assignment]
for exc in exception_tuple:
if not issubclass(exc, BaseException):
raise TypeError('"exception" must be an exception type.')
self._error_handlers[exc] = handler_callable
# ------------------------------------------------------------------------
# Helper methods
# ------------------------------------------------------------------------
def _schedule_callbacks(self, resp: Response) -> None:
callbacks = resp._registered_callbacks
# PERF(vytas): resp._registered_callbacks is already checked directly
# to shave off a function call since this is a hot/critical code path.
# if not callbacks:
# return
loop = asyncio.get_running_loop()
for cb, is_async in callbacks or ():
if is_async:
loop.create_task(cb()) # type: ignore[arg-type]
else:
loop.run_in_executor(None, cb)
async def _call_lifespan_handlers(
self, ver: str, scope: Dict[str, Any], receive: AsgiReceive, send: AsgiSend
) -> None:
while True:
event = await receive()
if event['type'] == 'lifespan.startup':
# PERF(vytas): Perform these sanity checks upon application
# startup, as opposed to repeating them every request.
# NOTE(vytas): If missing, 'asgi' is populated in __call__.
asgi_info: Dict[str, str] = scope['asgi']
version = asgi_info.get('version', '2.0 (implicit)')
if not version.startswith('3.'):
await send(
{
'type': EventType.LIFESPAN_STARTUP_FAILED,
'message': (
'Falcon requires ASGI version 3.x. '
f'Detected: {version}.'
),
}
)
return
if self.req_options._auto_parse_form_urlencoded:
await send(
{
'type': EventType.LIFESPAN_STARTUP_FAILED,
'message': (
'The deprecated WSGI RequestOptions.'
'auto_parse_form_urlencoded '
'option is not supported for Falcon ASGI apps. '
'Please use Request.get_media() instead. '
),
}
)
return
for handler in self._unprepared_middleware:
if hasattr(handler, 'process_startup'):
try:
await handler.process_startup(scope, event)
except Exception:
await send(
{
'type': EventType.LIFESPAN_STARTUP_FAILED,
'message': traceback.format_exc(),
}
)
return
await send({'type': EventType.LIFESPAN_STARTUP_COMPLETE})
elif event['type'] == 'lifespan.shutdown':
for handler in reversed(self._unprepared_middleware):
if hasattr(handler, 'process_shutdown'):
try:
await handler.process_shutdown(scope, event)
except Exception:
await send(
{
'type': EventType.LIFESPAN_SHUTDOWN_FAILED,
'message': traceback.format_exc(),
}
)
return
await send({'type': EventType.LIFESPAN_SHUTDOWN_COMPLETE})
return
async def _handle_websocket(
self, ver: str, scope: Dict[str, Any], receive: AsgiReceive, send: AsgiSend
) -> None:
first_event = await receive()
if first_event['type'] != EventType.WS_CONNECT:
# NOTE(kgriffs): The handshake was abandoned or this is a message
# we don't support, so bail out. This also fulfills the ASGI
# spec requirement to only process the request after
# receiving and verifying the first event.
response = {'type': EventType.WS_CLOSE, 'code': WSCloseCode.SERVER_ERROR}
if _supports_reason(ver):
response['reason'] = 'Internal Server Error'
await send(response)
return
req = self._request_type(scope, receive, options=self.req_options)
web_socket = WebSocket(
ver,
scope,
receive,
send,
self.ws_options.media_handlers,
self.ws_options.max_receive_queue,
self.ws_options.default_close_reasons,
)
params: Dict[str, Any] = {}
request_mw, resource_mw = self._middleware_ws
try:
for process_request_ws in request_mw:
await process_request_ws(req, web_socket)
on_websocket: AsgiResponderWsCallable
on_websocket, params, resource, req.uri_template = self._get_responder(req) # type: ignore[assignment]
# NOTE(kgriffs): If the request did not match any
# route, a default responder is returned and the
# resource is None. In that case, we skip the
# resource middleware methods. Resource will also be
# None when a middleware method already set
# resp.complete to True.
if resource:
for process_resource_ws in resource_mw:
await process_resource_ws(req, web_socket, resource, params)
await on_websocket(req, web_socket, **params)
await web_socket.close()
except Exception as ex:
if not await self._handle_exception(req, None, ex, params, ws=web_socket):
raise
def _prepare_middleware( # type: ignore[override]
self, middleware: List[object], independent_middleware: bool = False
) -> AsyncPreparedMiddlewareResult:
self._middleware_ws = prepare_middleware_ws(middleware)
return prepare_middleware(
middleware=middleware,
independent_middleware=independent_middleware,
asgi=True,
)
async def _http_status_handler( # type: ignore[override]
self,
req: Request,
resp: Optional[Response],
status: HTTPStatus,
params: Dict[str, Any],
ws: Optional[WebSocket] = None,
) -> None:
if resp:
self._compose_status_response(req, resp, status)
elif ws:
code = http_status_to_ws_code(status.status_code)
falcon._logger.error(
'[FALCON] HTTPStatus %s raised while handling WebSocket. '
'Closing with code %s',
status,
code,
)
await ws.close(code)
else:
raise NotImplementedError('resp or ws expected')
async def _http_error_handler( # type: ignore[override]
self,
req: Request,
resp: Optional[Response],
error: HTTPError,
params: Dict[str, Any],
ws: Optional[WebSocket] = None,
) -> None:
if resp:
self._compose_error_response(req, resp, error)
elif ws:
code = http_status_to_ws_code(error.status_code)
falcon._logger.error(
'[FALCON] HTTPError %s raised while handling WebSocket. '
'Closing with code %s',
error,
code,
)
await ws.close(code)
else:
raise NotImplementedError('resp or ws expected')
async def _python_error_handler( # type: ignore[override]
self,
req: Request,
resp: Optional[Response],
error: BaseException,
params: Dict[str, Any],
ws: Optional[WebSocket] = None,
) -> None:
falcon._logger.error('[FALCON] Unhandled exception in ASGI app', exc_info=error)
if resp:
self._compose_error_response(req, resp, falcon.HTTPInternalServerError())
elif ws:
await self._ws_cleanup_on_error(ws)
else:
raise NotImplementedError('resp or ws expected')
async def _ws_disconnected_error_handler(
self,
req: Request,
resp: Optional[Response],
error: WebSocketDisconnected,
params: Dict[str, Any],
ws: Optional[WebSocket] = None,
) -> None:
assert resp is None
assert ws is not None
falcon._logger.debug(
'[FALCON] WebSocket client disconnected with code %i', error.code
)
await self._ws_cleanup_on_error(ws)
if TYPE_CHECKING:
def _find_error_handler( # type: ignore[override]
self, ex: BaseException
) -> Optional[AsgiErrorHandler]: ...
async def _handle_exception( # type: ignore[override]
self,
req: Request,
resp: Optional[Response],
ex: BaseException,
params: Dict[str, Any],
ws: Optional[WebSocket] = None,
) -> bool:
"""Handle an exception raised from mw or a responder.
Args:
ex: Exception to handle
req: Current request object to pass to the handler registered for
the given exception type
resp: Current response object to pass to the handler registered for
the given exception type. Will be ``None`` in the case of a
WebSocket request.
params: Responder params to pass to the handler
registered for the given exception type
Keyword Args:
ws: WebSocket instance in the case that the error was raised while
handling a WebSocket connection.
Returns:
bool: ``True`` if a handler was found and called for the
exception, ``False`` otherwise.
"""
err_handler = self._find_error_handler(ex)
if resp:
# NOTE(caselit): Reset body, data and media before calling the handler
resp.text = resp.data = resp.media = None
if err_handler is not None:
try:
kwargs = {}
if ws and 'ws' in get_argnames(err_handler):
kwargs['ws'] = ws
await err_handler(req, resp, ex, params, **kwargs)
except HTTPStatus as status:
await self._http_status_handler(req, resp, status, params, ws=ws)
except HTTPError as error:
await self._http_error_handler(req, resp, error, params, ws=ws)
return True
# NOTE(kgriffs): No error handlers are defined for ex
# and it is not one of (HTTPStatus, HTTPError), since it
# would have matched one of the corresponding default
# handlers.
return False
async def _ws_cleanup_on_error(self, ws: WebSocket) -> None:
# NOTE(kgriffs): Attempt to close cleanly on our end
try:
await ws.close(self.ws_options.error_close_code)
except Exception as ex:
# NOTE(kgriffs): This can be raised by Daphne. We also
# may raise it ourselves for errors codes < 1000, but in that
# case we just include this string in the exception message
# to make it easier to verify test coverage of the following.
if 'invalid close code' in str(ex).lower():
await ws.close(_FALLBACK_WS_ERROR_CODE)
else:
falcon._logger.warning(
(
'[FALCON] Attempt to close web connection cleanly '
'failed due to raised error.'
),
exc_info=True,
)
raise