aiohttp middlewares¶
A middleware is a coroutine that can modify either the request or response. For example, here’s a simple middleware which appends 😃 to the response:
from aiohttp import web
@web.middleware
async def middleware(request: web.Request,
handler: Callable[[web.Request], Awaitable[web.Response]]):
resp = await handler(request)
resp.text = f"{resp.text} 😃"
return resp
The middleware is applied to all aiohttp application routes.
To setup middleware pass it to web.Application
constructor:
app = web.Application(middlewares=[middleware])
Let’s make something useful!
Error handling¶
@web.middleware
async def error_middleware(
request: web.Request,
handler: Callable[[web.Request], Awaitable[web.StreamResponse]],
) -> web.StreamResponse:
try:
return await handler(request)
except web.HTTPException:
raise
except asyncio.CancelledError:
raise
except Exception as ex:
return aiohttp_jinja2.render_template(
"error-page.html", request, {"error_text": str(ex)}, status=400
)
error-page.html
template:
{% block title %}
Site Error
{% endblock %}
{% block content %}
<h1>Bad thing happens</h1>
<p>
<b>
<big>
{{ error_text }}
</big>
</b>
</p>
{% endblock %}
Modify request¶
Rewrite protocol/host/remote-ip:
@web.middleware
async def middleware(request: web.Request,
handler: Callable[[web.Request], Awaitable[web.Response]]):
real_ip = request.headers['X-Forwarded-For']
real_host = request.headers['X-Forwarded-Host']
real_proto = request.headers['X-Forwarded-Proto']
new_request = request.clone(remote=real_ip,
host=real_host,
scheme=real_proto)
return await handler(request)
See also https://github.com/aio-libs/aiohttp-remotes for such middlewares.
DB Transaction handling¶
@web.middleware
async def middleware(request: web.Request,
handler: Callable[[web.Request], Awaitable[web.Response]]):
db = request.config_dict["DB"]
await db.execute("BEGIN")
try:
resp = await handler(request)
await db.execute("COMMIT")
return resp
except Exception:
await db.execute("ROLLBACK")
raise
Check for login¶
@web.middleware
async def middleware(request: web.Request,
handler: Callable[[web.Request], Awaitable[web.Response]]):
user = await get_user(request)
if user is not None:
request['USER'] = user
else: # user is None
if login_required(handler):
raise web.HTTPSeeAlso(location='/login')
return await handler(request)
Full example for server with error-page middleware¶
Example for HTML version of blogs server with images: Full server with error-page middleware