List of Middlewares#

As of version 1.0.0 aiohttp-middlewares library contains 5 middlewares to help aiohttp.web applications cover common web-development needs.

If some middleware is missed and should be added, feel free to create new feature request on GitHub.

Error Middleware#

New in version 0.2.0.

Middleware to handle errors in aiohttp applications.

Changed in version 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:

{
    "detail": "str"
}

For example, if view handler raises ValueError("wrong value") the default error handler provides 500 Server Error 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.

Usage#

import re

from aiohttp import web
from aiohttp_middlewares import (
    default_error_handler,
    error_context,
    error_middleware,
)


# Error handler for API requests
async def api_error(request: web.Request) -> web.Response:
    with error_context(request) as context:
        return web.json_response(
            context.data, 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(
    middlewares=[
        error_middleware(
            default_handler=default_error_handler,
            config={re.compile(r"^\/api"): api_error},
        )
    ]
)

# Ignore aiohttp.web HTTP Not Found errors from handling via middleware
app = web.Application(
    middlewares=[
        error_middleware(ignore_exceptions=web.HTTPNotFound)
    ]
)

CORS Middleware#

New in version 0.2.0.

Dealing with CORS headers for aiohttp applications.

IMPORTANT: There is a aiohttp-cors library, which handles CORS headers by attaching additional handlers to aiohttp application for OPTIONS (preflight) requests. In same time this CORS middleware mimics the logic of django-cors-headers, where all handling done in the middleware without any additional handlers. This approach allows aiohttp application to respond with CORS headers for OPTIONS or wildcard handlers, which is not possible with aiohttp-cors due to https://github.com/aio-libs/aiohttp-cors/issues/241 issue.

For detailed information about CORS (Cross Origin Resource Sharing) please visit:

Wikipedia
Or MDN

Changed in version 2.1.0.

If CORS request ends up with aiohttp.web.HTTPException - the exception will be used as a handler response and CORS headers will still be added to it, which means that even aiohttp cannot properly handle request to some URL, the response will still contain all necessary CORS headers.

Note

Thanks Kaarle Ritvanen for fix in aiohttp-middlewares#98 and Jonathan Heathcote for the review.

Configuration#

IMPORTANT: By default, CORS middleware do not allow any origins to access content from your aiohttp appliction. Which means, you need carefully check possible options and provide custom values for your needs.

Usage#

import re

from aiohttp import web
from aiohttp_middlewares import cors_middleware
from aiohttp_middlewares.cors import DEFAULT_ALLOW_HEADERS

# Unsecure configuration to allow all CORS requests
app = web.Application(
    middlewares=[cors_middleware(allow_all=True)]
)

# Allow CORS requests from URL http://localhost:3000
app = web.Application(
    middlewares=[
        cors_middleware(origins=["http://localhost:3000"])
    ]
)

# Allow CORS requests from all localhost urls
app = web.Application(
    middlewares=[
        cors_middleware(
            origins=[re.compile(r"^https?\:\/\/localhost")]
        )
    ]
)

# Allow CORS requests from https://frontend.myapp.com as well
# as allow credentials
CORS_ALLOW_ORIGINS = ["https://frontend.myapp.com"]
app = web.Application(
    middlewares=[
        cors_middleware(
            origins=CORS_ALLOW_ORIGINS,
            allow_credentials=True,
        )
    ]
)

# Allow CORS requests only for API urls
app = web.Application(
    middelwares=[
        cors_middleware(
            origins=CORS_ALLOW_ORIGINS,
            urls=[re.compile(r"^\/api")],
        )
    ]
)

# Allow CORS requests for POST & PATCH methods, and for all
# default headers and `X-Client-UID`
app = web.Application(
    middlewares=[
        cors_middleware(
            origins=CORS_ALLOW_ORIGINS,
            allow_methods=("POST", "PATCH"),
            allow_headers=DEFAULT_ALLOW_HEADERS
            + ("X-Client-UID",),
        )
    ]
)

Timeout Middleware#

Middleware to ensure that request handling does not exceeds X seconds.

Usage#

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

# Basic usage
app = web.Application(middlewares=[timeout_middleware(29.5)])

# Ignore slow responses from list of urls
slow_urls = (
    "/slow-url",
    "/very-slow-url",
    "/very/very/slow/url",
)
app = web.Application(
    middlewares=[timeout_middleware(4.5, ignore=slow_urls)]
)

# Ignore slow responsed from dict of urls. URL to ignore is a key,
# value is a lone string with HTTP method or list of strings with
# HTTP methods to ignore. HTTP methods are case-insensitive
slow_urls = {
    "/slow-url": "POST",
    "/very-slow-url": ("GET", "POST"),
}
app = web.Application(
    middlewares=[timeout_middleware(4.5, ignore=slow_urls)]
)

# Handle timeout errors with error middleware
app = web.Application(
    middlewares=[error_middleware(), timeout_middleware(14.5)]
)

Shield Middleware#

Middleware to shield application handlers by method or URL.

Usage#

from aiohttp import web
from aiohttp_middlewares import (
    NON_IDEMPOTENT_METHODS,
    shield_middleware,
)

# Basic usage (shield by handler method)
app = web.Application(
    middlewares=[shield_middleware(methods=IDEMPOTENT_METHODS)]
)

# Shield by handler URL
app = web.Application(
    middlewares=[shield_middleware(urls=["/", "/about-us"])]
)

# Shield by handler method, but ignore shielding list of URLs
app = web.Application(
    middlewares=[
        shield_middleware(
            methods=NON_IDEMPOTENT_METHODS,
            ignore={"/api/documents", "/api/comments"},
        )
    ]
)

# Combine shielding by method and URL
SHIELD_URLS = {
    "/api/documents": ["POST", "DELETE"],
    re.compile(r"/api/documents/\d+"): [
        "DELETE",
        "PUT",
        "PATCH",
    ],
}
app = web.Application(
    middlewares=[shield_middleware(urls=SHIELD_URLS)]
)

HTTPS Middleware#

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

Usage#

from aiohttp import web
from aiohttp_middlewares import https_middleware

# Basic usage
app = web.Application(middlewares=[https_middleware()])

# Specify custom headers to match, not `X-Forwarded-Proto: https`
app = web.Application(
    middlewares=https_middleware({"Forwarded": "https"})
)