Module

_middleware

Middleware extension system for server-level request/response processing.

Provides hooks for pre-request, post-response, and exception handling without modifying the ASGI bridge or requiring apps to wrap themselves in middleware.

Example:

async def auth_middleware(scope):
    '''Pre-request hook that can short-circuit.'''
    headers = dict(scope["headers"])
    if not headers.get(b"authorization"):
        return Response(status=401, body=b"Unauthorized")
    return scope  # Continue to app

async def cors_middleware(scope, status, headers):
    '''Post-response hook that modifies headers.'''
    headers.append((b"access-control-allow-origin", b"*"))
    return (status, headers)

config = ServerConfig(middleware=[auth_middleware, cors_middleware])
7Classes1Function

Classes

Response 3 ▼
Simple response object for middleware short-circuiting.

Simple response object for middleware short-circuiting.

Attributes

Name Type Description
status int —
headers list[tuple[bytes, bytes]] —
body bytes —
PreRequestMiddleware 1 ▼
Pre-request middleware hook. Called before the ASGI app with the request scope. Can inspect/modify…

Pre-request middleware hook.

Called before the ASGI app with the request scope. Can inspect/modify the scope or short-circuit by returning a Response.

Methods

Internal Methods 1 ▼
__call__ 1 dict[str, Any] | Response ▼
Process request before app.
async
async def __call__(self, scope: dict[str, Any]) -> dict[str, Any] | Response
Parameters
Name Type Description
scope —

ASGI scope dict
Returns
dict[str, Any] | Response Modified scope or Response to short-circuit
PostResponseMiddleware 1 ▼
Post-response middleware hook. Called after the app has processed the request but before the respo…

Post-response middleware hook.

Called after the app has processed the request but before the response is sent. Can modify status code or headers.

Methods

Internal Methods 1 ▼
__call__ 3 tuple[int, list[tuple[by… ▼
Process response after app.
async
async def __call__(self, scope: dict[str, Any], status: int, headers: list[tuple[bytes, bytes]]) -> tuple[int, list[tuple[bytes, bytes]]]
Parameters
Name Type Description
scope —

ASGI scope dict
status —

HTTP status code
headers —

Response headers
Returns
tuple[int, list[tuple[bytes, bytes]]] (status, headers) tuple
ExceptionMiddleware 1 ▼
Exception middleware hook. Called when the ASGI app raises an exception. Can return a custom respo…

Exception middleware hook.

Called when the ASGI app raises an exception. Can return a custom response or None to re-raise.

Methods

Internal Methods 1 ▼
__call__ 2 Response | None ▼
Handle exception from app.
async
async def __call__(self, scope: dict[str, Any], exc: Exception) -> Response | None
Parameters
Name Type Description
scope —

ASGI scope dict
exc —

Exception that was raised
Returns
Response | None Response to send, or None to re-raise
MiddlewareStack 3 ▼
Executes middleware hooks in order around an ASGI app call.

Executes middleware hooks in order around an ASGI app call.

Methods

Internal Methods 3 ▼
__init__ 2 ▼
def __init__(self, middleware: list[Middleware], app: Callable[[dict[str, Any], Receive, Send], Awaitable[None]]) -> None
Parameters
Name Type Description
middleware —
app —
__call__ 3 ▼
Execute middleware stack around app call. 1. Run pre-request middleware (can s…
async
async def __call__(self, scope: dict[str, Any], receive: Receive, send: Send) -> None

Execute middleware stack around app call.

  1. Run pre-request middleware (can short-circuit)
  2. If not short-circuited, call app
  3. Run post-response middleware (intercept first response.start)
  4. Run exception middleware if app raises
Parameters
Name Type Description
scope —
receive —
send —
_send_response 2 ▼
Send a Response object through ASGI send.
async
async def _send_response(self, response: Response, send: Send) -> None
Parameters
Name Type Description
response —

Response to send
send —

ASGI send callable
CORSMiddleware 2 ▼
CORS middleware that adds Access-Control headers. .. warning:: The default ``allow_origin="*"…

CORS middleware that adds Access-Control headers.

.. warning::

The default ``allow_origin="*"`` is suitable for development but
should be restricted to specific origins in production to prevent
cross-origin data leakage.

Methods

Internal Methods 2 ▼
__init__ 4 ▼
def __init__(self, allow_origin: str = '*', allow_methods: str = 'GET, POST, PUT, DELETE, OPTIONS', allow_headers: str = '*', max_age: int = 3600) -> None
Parameters
Name Type Description
allow_origin — Default:'*'
allow_methods — Default:'GET, POST, PUT, DELETE, OPTIONS'
allow_headers — Default:'*'
max_age — Default:3600
__call__ 3 tuple[int, list[tuple[by… ▼
Add CORS headers to response, skipping any already set by the app.
async
async def __call__(self, scope: dict[str, Any], status: int, headers: list[tuple[bytes, bytes]]) -> tuple[int, list[tuple[bytes, bytes]]]
Parameters
Name Type Description
scope —
status —
headers —
Returns
tuple[int, list[tuple[bytes, bytes]]]
SecurityHeadersMiddleware 2 ▼
Security headers middleware. Adds common security headers to all responses. Each header can be cu…

Security headers middleware.

Adds common security headers to all responses. Each header can be customised or suppressed (pass""to omit a header).

Default headers:

  • X-Frame-Options: DENY
  • X-Content-Type-Options: nosniff
  • X-XSS-Protection: 1; mode=block
  • Strict-Transport-Security(empty by default — pass an explicit value for production)
  • Content-Security-Policy: default-src 'self'
  • Referrer-Policy: strict-origin-when-cross-origin
  • Permissions-Policy: camera=(), microphone=(), geolocation=()

Methods

Internal Methods 2 ▼
__init__ 7 ▼
def __init__(self, *, x_frame_options: str = 'DENY', x_content_type_options: str = 'nosniff', x_xss_protection: str = '1; mode=block', hsts: str = '', csp: str = "default-src 'self'", referrer_policy: str = 'strict-origin-when-cross-origin', permissions_policy: str = 'camera=(), microphone=(), geolocation=()') -> None
Parameters
Name Type Description
x_frame_options — Default:'DENY'
x_content_type_options — Default:'nosniff'
x_xss_protection — Default:'1; mode=block'
hsts — Default:''
csp — Default:"default-src 'self'"
referrer_policy — Default:'strict-origin-when-cross-origin'
permissions_policy — Default:'camera=(), microphone=(), geolocation=()'
__call__ 3 tuple[int, list[tuple[by… ▼
Add security headers to response, skipping any already set by the app.
async
async def __call__(self, scope: dict[str, Any], status: int, headers: list[tuple[bytes, bytes]]) -> tuple[int, list[tuple[bytes, bytes]]]
Parameters
Name Type Description
scope —
status —
headers —
Returns
tuple[int, list[tuple[bytes, bytes]]]

Functions

_sanitize_headers 1 list[tuple[bytes, bytes]] ▼
Strip CR/LF characters from header names and values. Reuses the canonical sani…
def _sanitize_headers(headers: list[tuple[bytes, bytes]]) -> list[tuple[bytes, bytes]]

Strip CR/LF characters from header names and values.

Reuses the canonical sanitization applied in the ASGI bridge for app headers to avoid drift between duplicate implementations.

Parameters
Name Type Description
headers list[tuple[bytes, bytes]]
Returns
list[tuple[bytes, bytes]]