aiohttp Middleware: From Basics to Custom Implementation

Middleware in aiohttp allows you to intercept and modify requests or responses during the request-response cycle.

In this tutorial, you’ll learn how to implement middleware in aiohttp applications.

 

 

Request-response Cycle in aiohttp

Here’s a basic example of a request-response cycle without middleware:

from aiohttp import web
async def hello(request):
    return web.Response(text="Hello, aiohttp!")
app = web.Application()
app.router.add_get('/', hello)
web.run_app(app)

Output:

======== Running on http://0.0.0.0:8080 ========
(Press CTRL+C to quit)

This code starts a server that responds with “Hello, aiohttp!” when you access the root URL.

The request goes directly to the hello function, which returns a response text.

 

Middleware Execution Order

Middleware in aiohttp is executed in a specific order.

To show the middleware execution order, you can add multiple middleware to your application:

from aiohttp import web
async def middleware1(app, handler):
    async def middleware(request):
        print("Middleware 1: Before request")
        response = await handler(request)
        print("Middleware 1: After response")
        return response
    return middleware

async def middleware2(app, handler):
    async def middleware(request):
        print("Middleware 2: Before request")
        response = await handler(request)
        print("Middleware 2: After response")
        return response
    return middleware
async def hello(request):
    print("Handler: Processing request")
    return web.Response(text="Hello, aiohttp!")
app = web.Application(middlewares=[middleware1, middleware2])
app.router.add_get('/', hello)
web.run_app(app)

Output:

Middleware 1: Before request
Middleware 2: Before request
Handler: Processing request
Middleware 2: After response
Middleware 1: After response

The output shows that middleware is executed in the order it’s added with each middleware wrapping the subsequent ones and the handler.

This allows you to create a chain of operations that processes the request and response in a specific sequence.

You can chain multiple middleware together to create complex request processing.

 

Types of Middleware

Application middleware

Application middleware is applied to all routes in your application. This type is useful for global operations.

Here’s an example of application middleware that adds a custom header to all responses:

from aiohttp import web
import time
import asyncio
async def timing_middleware(app, handler):
    async def middleware(request):
        start = time.time()
        response = await handler(request)
        end = time.time()
        response.headers['X-Process-Time'] = f"{(end - start):.4f} seconds"
        return response
    return middleware
app = web.Application(middlewares=[timing_middleware])
async def hello(request):
    await asyncio.sleep(0.1)  # Simulate some processing
    return web.Response(text="Hello, aiohttp!")
app.router.add_get('/', hello)
web.run_app(app)

When you access the endpoint, the response will include an ‘X-Process-Time’ header showing the processing time.

Router Middleware

Router middleware is applied to specific routes or groups of routes.

It’s useful when you want to apply middleware to a subset of your application’s endpoints.

Here’s an example of router middleware that checks for authentication on admin routes:

from aiohttp import web
async def auth_middleware(request, handler):
    if request.path.startswith('/admin'):
        if 'Authorization' not in request.headers:
            raise web.HTTPUnauthorized(text='Authorization required')
    return await handler(request)
app = web.Application()
admin_routes = web.RouteTableDef()

@admin_routes.get('/admin')
async def admin(request):
    return web.Response(text="Admin page")
app.router.add_routes(admin_routes)
app.router.middleware.append(auth_middleware)
web.run_app(app)

This middleware checks for an Authorization header on all routes starting with ‘/admin’.

If the header is missing, it raises an HTTPUnauthorized exception.

This method is much better than creating a middleware and returning to each route in the code to check if you want to apply the route or not.

 

Create Custom Middleware

Function-based Middleware

Function-based middleware is the simplest form of middleware in aiohttp.

Here’s an example that logs request information:

from aiohttp import web
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def logging_middleware(app, handler):
    async def middleware(request):
        logger.info(f"Request: {request.method} {request.path}")
        response = await handler(request)
        logger.info(f"Response: {response.status}")
        return response
    return middleware
app = web.Application(middlewares=[logging_middleware])
async def hello(request):
    return web.Response(text="Hello, aiohttp!")
app.router.add_get('/', hello)
web.run_app(app)

Output:

INFO:__main__:Request: GET /
INFO:__main__:Response: 200
INFO:aiohttp.access:127.0.0.1 [08/Sep/2024:05:24:43 +0200] "GET / HTTP/1.1" 200 168 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0"

This middleware logs the request method, path, and response status for each request handled by the application.

Class-based Middleware

Class-based middleware allows you to create more complex middleware with state:

from aiohttp import web
class RateLimitMiddleware:
    def __init__(self, rate_limit=5):
        self.rate_limit = rate_limit
        self.request_count = {}
    async def middleware(self, app, handler):
        async def middleware_handler(request):
            client_ip = request.remote
            if client_ip in self.request_count:
                if self.request_count[client_ip] >= self.rate_limit:
                    raise web.HTTPTooManyRequests(text="Rate limit exceeded")
                self.request_count[client_ip] += 1
            else:
                self.request_count[client_ip] = 1
            response = await handler(request)
            return response
        return middleware_handler
app = web.Application(middlewares=[RateLimitMiddleware(rate_limit=3).middleware])
async def hello(request):
    return web.Response(text="Hello, aiohttp!")
app.router.add_get('/', hello)
web.run_app(app)

This middleware implements a simple rate-limiting mechanism that allows only 3 requests per client IP address.

After the limit is reached, it raises an HTTPTooManyRequests exception.