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.'''
    if not scope["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])
7Classes

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.

CORS middleware that adds Access-Control headers.

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.
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 1 ▼
Security headers middleware. Adds common security headers to all responses: - X-Frame-Options: DEN…

Security headers middleware.

Adds common security headers to all responses:

  • X-Frame-Options: DENY
  • X-Content-Type-Options: nosniff
  • X-XSS-Protection: 1; mode=block

Methods

Internal Methods 1 ▼
__call__ 3 tuple[int, list[tuple[by… ▼
Add security headers to response.
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]]]