Collection of useful middlewares for aiohttp applications.

  • Works on Python 3.5+
  • BSD licensed
  • Source, issues, and pull requests on GitHub


pip install aiohttp-middlewares


aiohttp-middlewares is licensed under the terms of BSD License.



Collection of useful middlewares for aiohttp applications.


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


from aiohttp import web
from aiohttp_middlewares import error_middleware, timeout_middleware

# Basic usage
app = web.Application(

# 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)])
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.Timeout in case of exceeding seconds per request, but not handling the error by itself. If you need to handle this error, please place error_middleware_factory 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.

  • seconds (Union[int, float]) – Max amount of seconds for each handler call.
  • ignore (Union[List[Union[str, Pattern[AnyStr]]], Set[Union[str, Pattern[AnyStr]]], Tuple[Union[str, Pattern[AnyStr]], …], Dict[Union[str, Pattern[AnyStr]], Union[List[str], FrozenSet[str], Set[str], Tuple[str, …], 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[[Application, Callable[[Request], Response]], Awaitable[Callable[[Request], Response]]]


Middleware to shield application handlers by method or URL.


from aiohttp import web
from aiohttp_middlewares import NON_IDEMPOTENT_METHODS, shield_middleware

# Basic usage (shield by handler method)
app = web.Application(

# 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(
            ignore={'/api/documents', '/api/comments'})])

# Combine shielding by method and URL
    '/api/documents': ['POST', 'DELETE'],
    re.compile('/api/documents/\d+'): ['DELETE', 'PUT', 'PATCH'],
app = web.Application(
aiohttp_middlewares.shield.shield_middleware(*, methods=None, urls=None, ignore=None)[source]

Ensure that handler execution would not break on CancelledError.

Shielding handlers allow to avoid breaking handler execution on 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 about shielding coroutines in official Python docs,

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. In Python 3.6+ you can supply standard dict here, in Python 3.5 please supply collections.OrderedDict instance instead.

To shield all non-idempotent methods you need to:

from aiohttp import web

app = web.Application(

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

import re

app = web.Application(
        re.compile(r'^/downloads/.*$'): 'GET',
        re.compile(r'.*'): NON_IDEMPOTENT_METHODS,
Return type:

Callable[[Application, Callable[[Request], Response]], Awaitable[Callable[[Request], Response]]]


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


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'}))

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

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[[Application, Callable[[Request], Response]], Awaitable[Callable[[Request], Response]]]


0.1.1 (2018-05-25)

  • Support async-timeout 3.0 version

0.1.0 (2018-02-20)

  • First non-beta release
  • Support aiohttp 3.0 version

0.1.0b2 (2018-02-04)

  • New shield_middleware to wrap request handler into asyncio.shield helper before execution
  • Allow to match URL by regexp for shield/timeout middleware

0.1.0b1 (2017-10-20)

  • New https_middleware to allow use proper scheme in request.url, when deploying aiohttp behind reverse proxy with enabled HTTPS
  • Allow passing dict of URLs with list methods to flex process of matching request ignored to wrapping into timeout context manager

0.1.0a2 (2017-05-14)

  • Rename timeout_middleware_factory to timeout_middleware

0.1.0a1 (2017-05-13)

  • Initial release. Implements timeout middleware