Middleware

ASGI middleware components.

Middleware package - ASGI middleware for genro-asgi.

class genro_asgi.middleware.BaseMiddleware(app, **kwargs)[source]

Bases: ABC

Base class for all middleware. Subclasses auto-register via __init_subclass__.

Class attributes:

middleware_name: Registry key (default: class name). middleware_order: Order in chain (lower = earlier). Ranges:

100: Core (errors) 200: Logging/Tracing 300: Security (cors, csrf) 400: Authentication (auth) 500-800: Business logic (custom) 900: Transformation (compression, caching)

middleware_default: Default on/off state. Default: False.

Use @headers_dict decorator on __call__ to access scope[“_headers”].

__init__(app, **kwargs)[source]

Initialize middleware with wrapped app.

Parameters:

app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – The ASGI app to wrap (next in chain).
**kwargs (Any) – Middleware options from the root.middleware(name={...}) entry in the config.py recipe.

app

middleware_default: bool = False

middleware_name: str = ''

middleware_order: int = 500

property server: Any

Walk the .app chain to find the AsgiServer via Dispatcher.

Returns None when used standalone (no Dispatcher in chain). Cached on first access.

genro_asgi.middleware.auth: alias of AuthMiddleware

genro_asgi.middleware.cache: alias of CacheMiddleware

genro_asgi.middleware.compression: alias of CompressionMiddleware

genro_asgi.middleware.cors: alias of CORSMiddleware

genro_asgi.middleware.errors: alias of ErrorMiddleware

genro_asgi.middleware.headers_dict(func)[source]

Decorator that parses headers into scope[“_headers”] dict if not present.

Return type:: Callable[..., Coroutine[Any, Any, None]]

genro_asgi.middleware.logging: alias of LoggingMiddleware

genro_asgi.middleware.middleware_chain(middleware_config, app)[source]

Build the middleware chain from the config.py recipe.

Uses middleware_order class attribute for sorting (lower = earlier in chain). Uses middleware_default class attribute for default on/off state.

Each entry’s value says both whether the middleware is on and how it is configured: a bool/string toggles it on the defaults; a dict turns it on and is passed as the middleware’s constructor kwargs:

root.middleware(cors=True)                            # on, defaults
root.middleware(cors={"allow_origins": ["x"]})        # on, with options
root.middleware(compression=False)                    # off

Parameters:

middleware_config (str | list[str] | dict[str, Any]) – Mapping {name: bool | dict} (a dict value carries the middleware options), a comma-separated string of names, or a list of names.
app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – The innermost ASGI app (usually the Dispatcher).

Return type:

Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]

Returns:

Wrapped ASGI app with middleware chain.

genro_asgi.middleware.session: alias of SessionMiddleware

genro_asgi.middleware.wellknown: alias of WellKnownMiddleware

Authentication

Authentication middleware for ASGI applications.

Supports Bearer tokens, Basic auth, and JWT with O(1) lookup at request time. Sets scope[“auth”] with authentication result for downstream handlers. The backend logic (config + verification) lives in authentication.AuthCore.

Backends:: bearer: Static token lookup. O(1) via dict. basic: Username/password. O(1) via base64-encoded key. jwt: Token verification via pyjwt. Falls back from bearer if not found.
Config:: bearer: Dict of {name: {token: “…”, tags: “…”}} basic: Dict of {username: {password: “…”, tags: “…”}} jwt: Dict of {name: {secret: “…”, algorithm: “…”, tags: “…”}}
scope[“auth”] format:: {“tags”: […], “identity”: “…”, “backend”: “bearer|basic|jwt:name”} None if no Authorization header present.

raises HTTPException(401):: If credentials present but invalid/expired.

Example

Configure authentication in the config.py recipe:

root.authMiddleware(
    bearer={"api_key": {"token": "sk_live_abc123", "tags": "api,read"}},
    basic={"admin": {"password": "secret", "tags": "admin"}},
    jwt={"internal": {"secret": "my-jwt-secret", "algorithm": "HS256"}},
)

class genro_asgi.middleware.authentication.AuthMiddleware(app, **entries)[source]

Bases: AuthCore, BaseMiddleware

Authentication middleware with O(1) credential lookup.

Extracts Authorization header, validates credentials against the AuthCore backends, and sets scope[“auth”] with result.

_auth_config: Dict mapping auth type to credentials dict.

Class Attributes:: middleware_name: “auth” - identifier for config. middleware_order: 400 - runs after CORS. middleware_default: False - disabled by default.

__init__(app, **entries)[source]

Initialize authentication middleware.

Parameters:

app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – Next ASGI application in the middleware chain.
**entries (Any) – Auth configuration by type (bearer, basic, jwt).

Note

Configuration is processed by AuthCore._configure_{type} methods. Unknown auth types are silently ignored.

middleware_default: bool = False

middleware_name: str = 'auth'

middleware_order: int = 400

Cache

Cache Middleware - HTTP caching headers for static files.

Adds cache-related headers to responses: - ETag: Based on file mtime + size for conditional requests - Last-Modified: From file modification time - Cache-Control: Configurable caching policy

Handles conditional requests: - If-None-Match: Returns 304 if ETag matches - If-Modified-Since: Returns 304 if file not modified

Config:: max_age (int): Cache-Control max-age in seconds. Default: 3600 (1 hour). immutable (bool): Add immutable directive for hashed filenames. Default: False. public (bool): Add public directive. Default: True.

Note

Requires scope[“_file_path”] to be set by the dispatcher for file responses. Only applies to GET/HEAD requests. Other methods pass through unchanged.

Example

Enable in the config.py recipe:

root.middleware(cache=True)

class genro_asgi.middleware.cache.CacheMiddleware(app, max_age=3600, immutable=False, public=True, **kwargs)[source]

Bases: BaseMiddleware

Cache middleware for static file responses.

Intercepts HTTP responses and adds caching headers (ETag, Last-Modified, Cache-Control). Handles conditional requests for 304 Not Modified responses.

max_age: Cache-Control max-age value in seconds.

immutable: Whether to add immutable directive.

public: Whether to add public directive.

Class Attributes:: middleware_name: “cache” - identifier for config. middleware_order: 900 - runs late to add headers to final response. middleware_default: False - disabled by default.

__init__(app, max_age=3600, immutable=False, public=True, **kwargs)[source]

Initialize cache middleware.

Parameters:

app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – Next ASGI application in the middleware chain.
max_age (int) – Cache-Control max-age in seconds. Defaults to 3600.
immutable (bool) – Add immutable directive for versioned assets. Defaults to False.
public (bool) – Add public directive allowing proxy caching. Defaults to True.
**kwargs (Any) – Additional arguments passed to BaseMiddleware.

immutable

max_age

middleware_default: bool = False

middleware_name: str = 'cache'

middleware_order: int = 900

public

CORS Middleware

CORS (Cross-Origin Resource Sharing) middleware for ASGI applications.

Adds CORS headers to HTTP responses, enabling cross-origin requests from browsers. Handles preflight OPTIONS requests automatically.

Config:: allow_origins (list|str): Origins allowed. Default: [“*”] allow_methods (list|str): HTTP methods allowed. Default: common methods allow_headers (list|str): Request headers allowed. Default: [“*”] allow_credentials (bool): Allow credentials (cookies). Default: False expose_headers (list|str): Response headers to expose. Default: [] max_age (int): Preflight cache time in seconds. Default: 600

Note

When allow_credentials is True, cannot use “*” for origins - the actual origin is echoed back instead.

Example

Enable CORS in the config.py recipe:

root.middleware(cors=True)

class genro_asgi.middleware.cors.CORSMiddleware(app, allow_origins=None, allow_methods=None, allow_headers=None, allow_credentials=False, expose_headers=None, max_age=600, **kwargs)[source]

Bases: BaseMiddleware

CORS middleware for HTTP requests.

Handles preflight OPTIONS requests and adds CORS headers to responses. Non-HTTP requests pass through unchanged.

allow_origins: List of allowed origins.

allow_methods: List of allowed HTTP methods.

allow_headers: List of allowed request headers.

allow_credentials: Whether to allow credentials.

expose_headers: List of headers to expose to browser.

max_age: Preflight response cache time in seconds.

Class Attributes:: middleware_name: “cors” - identifier for config. middleware_order: 300 - runs before auth middleware (outer wrapper). middleware_default: False - disabled by default.

__init__(app, allow_origins=None, allow_methods=None, allow_headers=None, allow_credentials=False, expose_headers=None, max_age=600, **kwargs)[source]

Initialize CORS middleware.

Parameters:

app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – Next ASGI application in the middleware chain.
allow_origins (str | list[str] | None) – Origins to allow. Accepts list or comma-separated string. Use “*” to allow all origins. Defaults to [“*”].
allow_methods (str | list[str] | None) – HTTP methods to allow. Defaults to common methods.
allow_headers (str | list[str] | None) – Request headers to allow. Defaults to [“*”].
allow_credentials (bool) – Allow cookies/auth headers. Defaults to False.
expose_headers (str | list[str] | None) – Response headers to expose to browser. Defaults to [].
max_age (int) – Preflight cache time in seconds. Defaults to 600.
**kwargs (Any) – Additional arguments passed to BaseMiddleware.

allow_credentials

allow_headers

allow_methods

allow_origins

expose_headers

max_age

middleware_default: bool = False

middleware_name: str = 'cors'

middleware_order: int = 300

Error Handling

Error handling middleware for ASGI applications.

Catches exceptions raised during request processing and converts them to appropriate HTTP responses. This is also the single seam where a 401 becomes a response, so the login challenge negotiation lives here.

Exception handling:

Redirect: Returns 3xx redirect with Location header
HTTPException: Returns status code with detail message; a 401 is negotiated when the login surface is active (see below)
Exception: Returns 500 Internal Server Error

Challenge negotiation (only when the server has login=True):

A 401 is the point where the server asks the caller to authenticate. API-first is preserved — a programmatic caller keeps the bare 401 (with WWW-Authenticate) and now also gets a body {"login_url": ...} so an SPA can find the page. A browser NAVIGATION (an http GET whose Accept includes text/html) instead gets a 302 to the login page carrying the original path+query as a validated next. With the login surface off the 401 is emitted exactly as before (no body change, no redirect).

Config:

debug (bool): If True, include traceback in 500 responses. Default: False.

Note

This middleware is enabled by default (middleware_default=True) and runs early in the chain (middleware_order=100) to catch all errors.

Example

Middleware is auto-enabled, but can be configured in the config.py recipe:

root.middleware(errors={"debug": True})  # tracebacks in development

class genro_asgi.middleware.errors.ErrorMiddleware(app, debug=False, **kwargs)[source]

Bases: BaseMiddleware

Error handling middleware for HTTP requests.

Wraps the application and catches exceptions, converting them to appropriate HTTP error responses. Non-HTTP requests pass through unchanged.

debug: If True, include stack traces in 500 error responses.

Class Attributes:: middleware_name: “errors” - identifier for config. middleware_order: 100 - runs early to catch all errors. middleware_default: True - enabled by default.

__init__(app, debug=False, **kwargs)[source]

Initialize error middleware.

Parameters:

app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – Next ASGI application in the middleware chain.
debug (bool) – Show tracebacks in 500 responses. Defaults to False.
**kwargs (Any) – Additional arguments passed to BaseMiddleware.

debug

middleware_default: bool = True

middleware_name: str = 'errors'

middleware_order: int = 100

Compression

Compression middleware for ASGI applications.

Compresses HTTP responses using gzip when beneficial. Buffers the response to determine if compression is worthwhile before sending.

Compression criteria:

Client accepts gzip (Accept-Encoding header contains “gzip”)
Response size >= minimum_size
Content-Type is compressible (text/*, application/json, etc.)
Compressed size < original size

Config:

minimum_size (int): Minimum bytes before compressing. Default: 500. compression_level (int): Gzip level 1-9. Default: 6.

Note

Adds Content-Encoding: gzip and Vary: Accept-Encoding headers. Updates Content-Length to compressed size.

Example

Enable in the config.py recipe:

root.middleware(compression=True)

class genro_asgi.middleware.compression.CompressionMiddleware(app, minimum_size=500, compression_level=6, **kwargs)[source]

Bases: BaseMiddleware

Gzip compression middleware for HTTP responses.

Buffers responses and applies gzip compression when all criteria are met. Non-HTTP requests pass through unchanged.

minimum_size: Minimum response size in bytes to consider compression.

compression_level: Gzip compression level (1=fast, 9=best).

Class Attributes:: middleware_name: “compression” - identifier for config. middleware_order: 900 - runs late to compress final response. middleware_default: False - disabled by default.

__init__(app, minimum_size=500, compression_level=6, **kwargs)[source]

Initialize compression middleware.

Parameters:

app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – Next ASGI application in the middleware chain.
minimum_size (int) – Minimum response size to compress. Defaults to 500.
compression_level (int) – Gzip level 1-9. Clamped to valid range. Defaults to 6.
**kwargs (Any) – Additional arguments passed to BaseMiddleware.

compression_level

middleware_default: bool = False

middleware_name: str = 'compression'

middleware_order: int = 900

minimum_size

Logging

Logging Middleware - HTTP request/response access logging.

Logs incoming requests and outgoing responses with timing information. Uses Python’s standard logging module for output.

Log format:: Request: “<- GET /api/users from 192.168.1.1” Response: “-> GET /api/users 200 (12.5ms)” Error: “-> GET /api/users ERROR: … (12.5ms)”
Config:: logger_name (str): Logger name. Default: “genro_asgi.access”. level (str): Log level (DEBUG, INFO, WARNING, ERROR). Default: “INFO”. include_headers (bool): Include request headers in DEBUG log. Default: False. include_query (bool): Include query string in request log. Default: True.

Example

Enable in the config.py recipe:

root.middleware(logging=True)

class genro_asgi.middleware.logging.LoggingMiddleware(app, logger_name='genro_asgi.access', level='INFO', include_headers=False, include_query=True, **kwargs)[source]

Bases: BaseMiddleware

Access logging middleware for HTTP requests.

Logs request arrival and response completion with timing. Uses Python’s logging module, allowing integration with existing logging configuration.

logger: Python Logger instance for access logs.

level: Numeric log level (from logging module).

include_headers: Whether to log request headers (at DEBUG level).

include_query: Whether to include query string in request path.

Class Attributes:: middleware_name: “logging” - identifier for config. middleware_order: 200 - runs early to capture full request timing. middleware_default: False - disabled by default.

__init__(app, logger_name='genro_asgi.access', level='INFO', include_headers=False, include_query=True, **kwargs)[source]

Initialize logging middleware.

Parameters:

app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – Next ASGI application in the middleware chain.
logger_name (str) – Name for the Python logger. Defaults to “genro_asgi.access”.
level (str) – Log level string (DEBUG, INFO, etc.). Defaults to “INFO”.
include_headers (bool) – Log headers at DEBUG level. Defaults to False.
include_query (bool) – Include query string in path log. Defaults to True.
**kwargs (Any) – Additional arguments passed to BaseMiddleware.

include_headers

include_query

level

logger

middleware_default: bool = False

middleware_name: str = 'logging'

middleware_order: int = 200

Session

Session middleware — session management driven by the request cookie.

Extracts the session token from the request cookie (read from the ASGI scope, not from a request object), reconnects or creates the session, injects it into scope[“session”], and sets Set-Cookie on new sessions.

Like the other server middleware (authentication, cache), it works on the scope: it reads scope["_headers"] (populated by @headers_dict) and parses the Cookie header itself, so it does not depend on the request object having been created yet.

Config:: cookie_name: “session_id” (default) secure: false (default) samesite: “lax” (default)

class genro_asgi.middleware.session.SessionMiddleware(app, *, cookie_name='session_id', secure=False, samesite='lax', **kwargs)[source]

Bases: BaseMiddleware

Per-app session middleware. Manages session lifecycle via cookies.

__init__(app, *, cookie_name='session_id', secure=False, samesite='lax', **kwargs)[source]

Initialize session middleware.

Parameters:

app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – Next ASGI app in chain.
cookie_name (str) – Name of the session cookie.
secure (bool) – Set Secure flag on cookie.
samesite (str) – SameSite policy for cookie.
**kwargs (Any) – Additional config (ignored).

middleware_default: bool = False

middleware_name: str = 'session'

middleware_order: int = 450