Source code for aiohttp_middlewares.error

Error Middleware

.. versionadded:: 0.2.0

Middleware to handle errors in aiohttp applications.

.. versionchanged:: 1.0.0

Previously, ``error_middleware`` required ``default_handler`` to be passed
on initialization. However in **1.0.0** version ``aiohttp-middlewares`` ships
default error handler, which log exception traceback into
``aiohttp_middlewares.error`` logger and responds with given JSON:

.. code-block:: json

        "detail": "str"

For example, if view handler raises ``ValueError("wrong value")`` the default
error handler provides 500 Server Error JSON:

.. code-block:: json

        "detail": "wrong value"

In same time, it is still able to provide custom default error handler if you
need more control on error handling.

Other notable change in **1.0.0** version is allowing to ignore exception or
tuple of exceptions (as in ``try/catch`` block) from handling via middleware.
This might be helpful, when you don't want, for example, to have in Sentry
``web.HTTPNotFound`` and/or ``web.BadRequest`` errors.


.. code-block:: python

    import re

    from aiohttp import web
    from aiohttp_middlewares import (

    # Error handler for API requests
    async def api_error(request: web.Request) -> web.Response:
        with error_context(request) as context:
            return web.json_response(
      , status=context.status

    # Basic usage (default error handler for whole application)
    app = web.Application(middlewares=[error_middleware()])

    # Advanced usage (multiple error handlers for different
    # application parts)
    app = web.Application(
                config={re.compile(r"^\/api"): api_error},

    # Ignore aiohttp.web HTTP Not Found errors from handling via middleware
    app = web.Application(


import logging
from contextlib import contextmanager
from functools import partial
from typing import Dict, Iterator, Tuple, Union

import attr
from aiohttp import web

from aiohttp_middlewares.annotations import (
from aiohttp_middlewares.utils import match_path

DEFAULT_EXCEPTION = Exception("Unhandled aiohttp-middlewares exception.")

Config = Dict[Url, Handler]
logger = logging.getLogger(__name__)

@attr.dataclass(frozen=True, slots=True)
class ErrorContext:
    """Context with all necessary data about the error."""

    err: Exception
    message: str
    status: int
    data: DictStrAny

[docs]async def default_error_handler(request: web.Request) -> web.Response: """Default error handler to respond with JSON error details. If, for example, ``aiohttp.web`` view handler raises ``ValueError("wrong value")`` exception, default error handler will produce JSON response of 500 status with given content: .. code-block:: json { "detail": "wrong value" } And to see the whole exception traceback in logs you need to enable ``aiohttp_middlewares`` in logging config. .. versionadded:: 1.0.0 """ with error_context(request) as context: logger.error(context.message, exc_info=True) return web.json_response(, status=context.status)
[docs]@contextmanager def error_context(request: web.Request) -> Iterator[ErrorContext]: """Context manager to retrieve error data inside of error handler (view). The result instance will contain: - Error itself - Error message (by default: ``str(err)``) - Error status (by default: ``500``) - Error data dict (by default: ``{"detail": str(err)}``) """ err = get_error_from_request(request) message = getattr(err, "message", None) or str(err) data = getattr(err, "data", None) or {"detail": message} status = getattr(err, "status", None) or 500 yield ErrorContext(err=err, message=message, status=status, data=data)
[docs]def error_middleware( *, default_handler: Handler = default_error_handler, config: Union[Config, None] = None, ignore_exceptions: Union[ ExceptionType, Tuple[ExceptionType, ...], None ] = None, ) -> Middleware: """Middleware to handle exceptions in aiohttp applications. To catch all possible errors, please put this middleware on top of your ``middlewares`` list (**but after CORS middleware if it is used**) as: .. code-block:: python from aiohttp import web from aiohttp_middlewares import ( error_middleware, timeout_middleware, ) app = web.Application( midllewares=[error_middleware(...), timeout_middleware(...)] ) :param default_handler: Default handler to called on error catched by error middleware. :param config: When application requires multiple error handlers, provide mapping in format ``Dict[Url, Handler]``, where ``Url`` can be an exact string to match path or regex and ``Handler`` is a handler to be called when ``Url`` matches current request path if any. :param ignore_exceptions: Do not process given exceptions via error middleware. """ get_response = partial( get_error_response, default_handler=default_handler, config=config, ignore_exceptions=ignore_exceptions, ) @web.middleware async def middleware( request: web.Request, handler: Handler ) -> web.StreamResponse: try: return await handler(request) except Exception as err: return await get_response(request, err) return middleware
def get_error_from_request(request: web.Request) -> Exception: """Get previously stored error from request dict. Return default exception if nothing stored before. """ return request.get(REQUEST_ERROR_KEY) or DEFAULT_EXCEPTION def get_error_handler( request: web.Request, config: Union[Config, None] ) -> Union[Handler, None]: """Find error handler matching current request path if any.""" if not config: return None path = request.rel_url.path for item, handler in config.items(): if match_path(item, path): return handler return None
[docs]async def get_error_response( request: web.Request, err: Exception, *, default_handler: Handler = default_error_handler, config: Union[Config, None] = None, ignore_exceptions: Union[ ExceptionType, Tuple[ExceptionType, ...], None ] = None, ) -> web.StreamResponse: """Actual coroutine to get response for given request & error. .. versionadded:: 1.1.0 This is a cornerstone of error middleware and can be reused in attempt to overwrite error middleware logic. For example, when you need to post-process response and it may result in extra exceptions it is useful to make ``custom_error_middleware`` as follows, .. code-block:: python from aiohttp import web from aiohttp_middlewares import get_error_response from aiohttp_middlewares.annotations import Handler @web.middleware async def custom_error_middleware( request: web.Request, handler: Handler ) -> web.StreamResponse: try: response = await handler(request) post_process_response(response) except Exception as err: return await get_error_response(request, err) """ if ignore_exceptions and isinstance(err, ignore_exceptions): raise err set_error_to_request(request, err) error_handler = get_error_handler(request, config) or default_handler return await error_handler(request)
def set_error_to_request(request: web.Request, err: Exception) -> Exception: """Store catched error to request dict.""" request[REQUEST_ERROR_KEY] = err return err