API Reference#

Middlewares#

Error Middleware#

aiohttp_middlewares.error.error_middleware(*, default_handler=<function default_error_handler>, config=None, ignore_exceptions=None)[source]#

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:

from aiohttp import web
from aiohttp_middlewares import (
    error_middleware,
    timeout_middleware,
)

app = web.Application(
    midllewares=[error_middleware(...), timeout_middleware(...)]
)
Parameters:

  • default_handler (Callable[[Request], Awaitable[StreamResponse]]) – Default handler to called on error catched by error middleware.

  • config (Optional[Dict[Union[str, Pattern[str], URL], Callable[[Request], Awaitable[StreamResponse]]]]) – 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.

  • ignore_exceptions (Union[Type[Exception], Tuple[Type[Exception], ...], None]) – Do not process given exceptions via error middleware.

Return type:

Callable[[Request, Callable[[Request], Awaitable[StreamResponse]]], Awaitable[StreamResponse]]

CORS Middleware#

aiohttp_middlewares.cors.cors_middleware(*, allow_all=False, origins=None, urls=None, expose_headers=None, allow_headers=('accept', 'accept-encoding', 'authorization', 'content-type', 'dnt', 'origin', 'user-agent', 'x-csrftoken', 'x-requested-with'), allow_methods=('DELETE', 'GET', 'OPTIONS', 'PATCH', 'POST', 'PUT'), allow_credentials=False, max_age=None)[source]#

Middleware to provide CORS headers for aiohttp applications.

Parameters:

  • allow_all (bool) – When enabled, allow any Origin to access content from your aiohttp web application. Please be careful with enabling this option as it may result in security issues for your application. By default: False

  • origins (Optional[Collection[Union[str, Pattern[str], URL]]]) – Allow content access for given list of origins. Support supplying strings for exact origin match or regex instances. By default: None

  • urls (Optional[Collection[Union[str, Pattern[str], URL]]]) – Allow content access for given list of URLs in aiohttp application. By default: apply CORS headers for all URLs

  • expose_headers (Optional[Collection[str]]) – List of headers to be exposed with every CORS request. By default: None

  • allow_headers (Collection[str]) –

    List of allowed headers. By default:

    (
    "accept",
    "accept-encoding",
    "authorization",
    "content-type",
    "dnt",
    "origin",
    "user-agent",
    "x-csrftoken",
    "x-requested-with",
)

  • allow_methods (Collection[str]) –

    List of allowed methods. By default:

    ("DELETE", "GET", "OPTIONS", "PATCH", "POST", "PUT")

  • allow_credentials (bool) – When enabled apply allow credentials header in response, which results in sharing cookies on shared resources. Please be careful with allowing credentials for CORS requests. By default: False

  • max_age (Optional[int]) – Access control max age in seconds. By default: None

Return type:

Callable[[Request, Callable[[Request], Awaitable[StreamResponse]]], Awaitable[StreamResponse]]

Timeout Middleware#

aiohttp_middlewares.timeout.timeout_middleware(seconds, *, ignore=None)[source]#

Ensure that request handling does not exceed X seconds.

This is helpful when aiohttp application served behind nginx or other reverse proxy with enabled read timeout. And when this read timeout exceeds reverse proxy generates error page instead of aiohttp app, which may result in bad user experience.

For best results, please do not supply seconds value which equals read timeout value at reverse proxy as it may results that request handling at aiohttp will be ended after reverse proxy already responded with 504 error. Timeout context manager accepts floats, so if nginx has read timeout in 30 seconds, it’s ok to configure timeout middleware to raise timeout error after 29.5 seconds. In that case in most cases user for sure will see the error from aiohttp app instead of reverse proxy.

Notice that timeout middleware just raised asyncio.TimeoutError in case of exceeding seconds per request, but not handling the error by itself. If you need to handle this error, please place aiohttp_middlewares.error.error_middleware() in list of application middlewares as well. Error middleware should be placed before timeout middleware, so timeout errors can be catched and processed properly.

In case if you need to “disable” timeout middleware for given request path, please supply ignore collection as second positional argument, like:

from aiohttp import web

app = web.Application(
    middlewares=[timeout_middleware(14.5, ignore={"/slow-url"})]
)

In case if you need more flexible ignore rules you can pass ignore dict, where key is an URL to ignore and value is a collection of methods to ignore from timeout handling for given URL.

ignore = {"/slow-url": ["POST"]}
app = web.Application(
    middlewares=[timeout_middleware(14.5, ignore=ignore)]
)

Behind the scene, when current request path match the URL from ignore collection or dict timeout context manager will be configured to avoid break the execution after X seconds.

Parameters:

  • seconds (Union[int, float]) – Max amount of seconds for each handler call.

  • ignore (Union[Collection[Union[str, Pattern[str], URL]], Dict[Union[str, Pattern[str], URL], Collection[str]], None]) – Do not limit execution for any of given URLs (paths). This is useful when request handler returns StreamResponse instead of regular Response. You also can specify URLs as dict, where key is URL to ignore from wrapping into timeout context and value is list of methods to ignore. This is helpful when you need ignore only POST requests of slow API endpoint, but still need to have GET requests to same endpoint to not exceed X seconds.

Return type:

Callable[[Request, Callable[[Request], Awaitable[StreamResponse]]], Awaitable[StreamResponse]]

Shield Middleware#

aiohttp_middlewares.shield.shield_middleware(*, methods=None, urls=None, ignore=None)[source]#

Ensure that handler execution would not break on asyncio.CancelledError.

Shielding handlers allow to avoid breaking handler execution on asyncio.CancelledError (this happens for example while client closes conneciton, but server still not ready to fullify response).

In most cases you need to shield non-idempotent methods (POST, PUT, PATCH, DELETE) and ignore shielding idempotent GET, HEAD, OPTIONS and TRACE requests.

More details about shielding coroutines in official Python docs: asyncio.shield()

Other possibility to allow shielding request handlers by URLs dict. In that case order of dict keys is necessary as they will be processed from first to last added.

To shield all non-idempotent methods you need to:

from aiohttp import web

app = web.Application(
    middlewares=[
        shield_middleware(methods=NON_IDEMPOTENT_METHODS)
    ]
)

To shield all non-idempotent methods and GET requests to /downloads/* URLs:

import re

app = web.Application(
    middlewares=[
        shield_middleware(
            urls={
                re.compile(r"^/downloads/.*$"): "GET",
                re.compile(r".*"): NON_IDEMPOTENT_METHODS,
            }
        )
    ]
)
Parameters:
Return type:

Callable[[Request, Callable[[Request], Awaitable[StreamResponse]]], Awaitable[StreamResponse]]

HTTPS Middleware#

aiohttp_middlewares.https.https_middleware(match_headers=None)[source]#

Change scheme for current request when aiohttp application deployed behind reverse proxy with HTTPS enabled.

This middleware is required to use, when your aiohttp app deployed behind nginx with HTTPS enabled, after aiohttp discounted secure_proxy_ssl_header keyword argument in https://github.com/aio-libs/aiohttp/pull/2299.

Parameters:

match_headers (Optional[Dict[str, str]]) –

Dict of header(s) from reverse proxy to specify that aiohttp run behind HTTPS. By default:

{"X-Forwarded-Proto": "https"}

Return type:

Callable[[Request, Callable[[Request], Awaitable[StreamResponse]]], Awaitable[StreamResponse]]

Other#

default_error_handler#

async aiohttp_middlewares.error.default_error_handler(request)[source]#

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:

{
    "detail": "wrong value"
}

And to see the whole exception traceback in logs you need to enable aiohttp_middlewares in logging config. :rtype: Response

New in version 1.0.0.

error_context#

aiohttp_middlewares.error.error_context(request)[source]#

Context manager to retrieve error data inside of error handler (view).

The result instance will contain: :rtype: Iterator[ErrorContext]

  • Error itself

  • Error message (by default: str(err))

  • Error status (by default: 500)

  • Error data dict (by default: {"detail": str(err)})

get_error_response#

async aiohttp_middlewares.error.get_error_response(request, err, *, default_handler=<function default_error_handler>, config=None, ignore_exceptions=None)[source]#

Actual coroutine to get response for given request & error. :rtype: StreamResponse

New in version 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,

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)

match_path#

aiohttp_middlewares.utils.match_path(item, path)[source]#

Check whether current path is equal to given URL str or regexp.

Parameters:
Return type:

bool