TopolearnCross-cutting concerns belong in one place instead of being repeated in every handler.
A middleware is a coroutine that wraps the handler chain. It receives the request and the next handler, and can act before and after calling it:
@web.middleware
async def timing(request, handler):
start = time.perf_counter()
response = await handler(request)
response.headers["X-Time"] = f"{time.perf_counter() - start:.4f}"
return responseRegister middlewares when creating the app: web.Application(middlewares=[error_mw, auth_mw, timing]). They run in list order on the way in, and in reverse order on the way out, like layers of an onion. The first middleware sees everything, including errors from the ones after it.
Common uses:
- Error handling: catch
web.HTTPExceptionor unexpected exceptions and render consistent JSON errors. Note thatHTTPExceptionis raised, so a middleware must catch it (or let it propagate to aiohttp's default handling). - Authentication: check a token, attach the user to the request (
request["user"] = ...) and either continue or raiseweb.HTTPUnauthorized. - Logging, metrics, CORS, request IDs.
A middleware can short-circuit by returning a response without calling handler. The request object supports item assignment (request["key"]), which is the standard way to pass data from middlewares to handlers.
For per-response modifications that do not need to wrap the call, the on_response_prepare signal is an alternative.
Example
import asyncio
from aiohttp import web
from aiohttp.test_utils import TestClient, TestServer
@web.middleware
async def errors_as_json(request, handler):
try:
return await handler(request)
except web.HTTPException as e:
return web.json_response({"error": e.reason}, status=e.status)
@web.middleware
async def auth(request, handler):
if request.headers.get("Authorization") != "Bearer secret":
raise web.HTTPUnauthorized()
request["user"] = "ada" # pass data to handlers
return await handler(request)
async def me(request):
return web.json_response({"user": request["user"]})
async def demo():
app = web.Application(middlewares=[errors_as_json, auth])
app.router.add_get("/me", me)
async with TestClient(TestServer(app)) as c:
for h in ({}, {"Authorization": "Bearer secret"}):
print(await (await c.get("/me", headers=h)).json()) # {'error': 'Unauthorized'}, then {'user': 'ada'}
asyncio.run(demo())Remember
- A middleware is async (request, handler) -> response wrapping the chain
- They run in list order going in, reverse order coming out
- Use request['key'] to pass data from middleware to handlers
Watch out for
- Putting the error-handling middleware last, so it can't catch errors raised by earlier middlewares.