Middlewares¶
Middlewares enable modifying the chain of functions that handle each web request.
This page covers:
- Introduction to middlewares.
- How to use function decorators to avoid code repetition.
- Middleware management with MiddlewareList and MiddlewareCategory.
- Organizing middlewares by categories and priorities.
Introduction to middlewares¶
Middlewares enable the definition of callbacks that are executed for each web request in a specific order.
Info
If a function should only be called for specific routes, use a decorator function instead.
Middlewares are executed in order: each receives the Request object as the
first parameter and the next handler to be called as the second parameter. Any
middleware can choose not to call the next handler and instead return a
Response object. For instance, a middleware can be used to return an HTTP
401 Unauthorized response in certain scenarios.
from blacksheep import Application, get
app = Application()
async def middleware_one(request, handler):
print("middleware 1: A")
response = await handler(request)
print("middleware 1: B")
return response
async def middleware_two(request, handler):
print("middleware 2: C")
response = await handler(request)
print("middleware 2: D")
return response
app.middlewares.append(middleware_one)
app.middlewares.append(middleware_two)
@get("/")
def home():
return "OK"
In this example, the following data would be printed to the console:
Middlewares defined as classes¶
To define a middleware as a class, make the class async callable, like in the example below:
class ExampleMiddleware:
async def __call__(self, request, handler):
# do something before passing the request to the next handler
response = await handler(request)
# do something after the following request handlers prepared the response
return response
The same example including type annotations:
from typing import Awaitable, Callable
from blacksheep import Request, Response
class ExampleMiddleware:
async def __call__(
self, request: Request, handler: Callable[[Request], Awaitable[Response]]
) -> Response:
# do something before passing the request to the next handler
response = await handler(request)
# do something after the following request handlers prepared the response
return response
Resolution chains¶
When middlewares are defined for an application, resolution chains are built at its start. Every handler configured in the application router is replaced by a chain, executing middlewares in order, down to the registered handler.
Middleware management with MiddlewareList and MiddlewareCategory¶
New in BlackSheep 2.4.4
Starting from BlackSheep 2.4.4, middleware management has been enhanced with MiddlewareList and MiddlewareCategory to simplify middleware ordering and organization.
The MiddlewareList is a specialized container that provides better control over middleware ordering through categories and priorities. This addresses common issues where middleware order matters significantly, such as ensuring authentication happens before authorization, or that CORS headers are set early in the pipeline.
Middleware categories¶
The MiddlewareCategory enum defines predefined categories that represent the typical order middlewares should be executed:
from blacksheep.middlewares import MiddlewareCategory
# Available categories (in execution order):
MiddlewareCategory.INIT # 10 - CORS, security headers, early configuration
MiddlewareCategory.SESSION # 20 - Session handling
MiddlewareCategory.AUTH # 30 - Authentication
MiddlewareCategory.AUTHZ # 40 - Authorization
MiddlewareCategory.BUSINESS # 50 - User business logic middlewares (default)
MiddlewareCategory.MESSAGE # 60 - Request/Response modification
Adding categorized middlewares¶
You can now specify a category and priority when adding middlewares:
from blacksheep import Application
from blacksheep.middlewares import MiddlewareCategory
app = Application()
# Add middleware with category and priority
app.middlewares.append(
cors_middleware,
category=MiddlewareCategory.INIT,
priority=0 # Lower priority = executed first within category
)
app.middlewares.append(
auth_middleware,
category=MiddlewareCategory.AUTH,
priority=0
)
app.middlewares.append(
custom_business_logic,
category=MiddlewareCategory.BUSINESS,
priority=10
)
# If no category is specified, defaults to BUSINESS
app.middlewares.append(logging_middleware)
Priority within categories¶
Within each category, middlewares are ordered by their priority value (lower values execute first):
# These will execute in order: middleware_a, middleware_b, middleware_c
app.middlewares.append(middleware_a, MiddlewareCategory.AUTH, priority=0)
app.middlewares.append(middleware_b, MiddlewareCategory.AUTH, priority=5)
app.middlewares.append(middleware_c, MiddlewareCategory.AUTH, priority=10)
Backward compatibility¶
The traditional append() and insert() methods continue to work:
# Traditional approach (still supported)
app.middlewares.append(my_middleware)
# Insert at specific position (defaults to INIT category for backward compatibility)
app.middlewares.insert(0, early_middleware)
Example: Complete middleware setup¶
Here's a comprehensive example showing how to organize middlewares by category:
from blacksheep import Application
from blacksheep.middlewares import MiddlewareCategory
from blacksheep.server.cors import CORSMiddleware
from blacksheep.server.authentication import AuthenticationMiddleware
from blacksheep.server.authorization import AuthorizationMiddleware
app = Application()
# CORS and security headers (execute first)
app.middlewares.append(
CORSMiddleware(),
category=MiddlewareCategory.INIT,
priority=0
)
# Session handling
app.middlewares.append(
session_middleware,
category=MiddlewareCategory.SESSION,
priority=0
)
# Authentication (after sessions)
app.middlewares.append(
AuthenticationMiddleware(),
category=MiddlewareCategory.AUTH,
priority=0
)
# Authorization (after authentication)
app.middlewares.append(
AuthorizationMiddleware(),
category=MiddlewareCategory.AUTHZ,
priority=0
)
# Custom business logic
app.middlewares.append(
rate_limiting_middleware,
category=MiddlewareCategory.BUSINESS,
priority=0
)
app.middlewares.append(
custom_logging_middleware,
category=MiddlewareCategory.BUSINESS,
priority=10
)
# Response modification (execute last)
app.middlewares.append(
response_time_middleware,
category=MiddlewareCategory.MESSAGE,
priority=0
)
Benefits of categorized middlewares¶
- Predictable ordering: Middlewares execute in a logical, predictable order based on their category.
- Easier maintenance: You can add middlewares without worrying about their position in a flat list.
- Better organization: Categories make it clear what each middleware's purpose is.
- Flexible priorities: Fine-tune execution order within categories using priority values.
- Backward compatibility: Existing code continues to work without changes.
Wrapping request handlers¶
When a common portion of logic should be applied to certain request handlers, but not to all of them, it is recommended to define a decorator.
The following example shows how to define a decorator that applies certain response headers only for certain routes.
from functools import wraps
from typing import Tuple
from blacksheep.server.normalization import ensure_response
def headers(additional_headers: Tuple[Tuple[str, str], ...]):
def decorator(next_handler):
@wraps(next_handler)
async def wrapped(*args, **kwargs) -> Response:
response = ensure_response(await next_handler(*args, **kwargs))
for (name, value) in additional_headers:
response.add_header(name.encode(), value.encode())
return response
return wrapped
return decorator
Then use the decorator on specific request handlers:
The order of decorators matters.
User-defined decorators must be applied before the route decorator (in the example above, before @get).
Define a wrapper compatible with synchronous and asynchronous functions¶
To define a wrapper that is compatible with both synchronous and asynchronous
functions, it is possible to use inspect.iscoroutinefunction function. For
example, to alter the decorator above to be also compatible with request
handlers defined as synchronous functions:
import inspect
from functools import wraps
from typing import Tuple
from blacksheep.server.normalization import ensure_response
def headers(additional_headers: Tuple[Tuple[str, str], ...]):
def decorator(next_handler):
if inspect.iscoroutinefunction(next_handler):
@wraps(next_handler)
async def wrapped(*args, **kwargs):
response = ensure_response(await next_handler(*args, **kwargs))
for (name, value) in additional_headers:
response.add_header(name.encode(), value.encode())
return response
return wrapped
else:
@wraps(next_handler)
def wrapped(*args, **kwargs):
response = ensure_response(next_handler(*args, **kwargs))
for (name, value) in additional_headers:
response.add_header(name.encode(), value.encode())
return response
return wrapped
return decorator
Warning
The ensure_response function is necessary to support scenarios
when the request handlers defined by the user doesn't return an instance of
Response class (see request handlers normalization).
Last modified on: 2025-12-13 23:01:35