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-specific configuration from YAML.

app

middleware_default: bool = False

middleware_name: str = ''

middleware_order: int = 500

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, full_config=None)[source]

Build middleware chain from config with automatic ordering.

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

YAML format:

middleware:

cors: on auth: on errors: on # default=True, so usually omitted

cors_middleware:

allow_origins: “*”

auth_middleware:

bearer:

reader_token:: token: “tk_abc123” tags: “read”

basic:

admin:: password: “secret” tags: “admin”

Parameters:

middleware_config (str | list[str] | dict[str, Any]) – Dict {name: on/off}, comma-separated string, or list.
app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – The innermost ASGI app (usually Dispatcher).
full_config (Any) – Full config object to lookup {name}_middleware sections.

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.

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.

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

Enable in config.yaml:

middleware:
  auth:
    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: BaseMiddleware

Authentication middleware with O(1) credential lookup.

Extracts Authorization header, validates credentials against configured 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 _configure_{type} methods. Unknown auth types are silently ignored.

middleware_default: bool = False

middleware_name: str = 'auth'

middleware_order: int = 400

verify_credentials(username, password)[source]

Verify username/password against basic auth config.

Useful for login endpoints that need to verify credentials before issuing a JWT token.

Parameters:

username (str) – The username to verify.
password (str) – The password to verify.

Return type:

dict[str, Any] | None

Returns:

Dict with tags and identity if valid, None otherwise.

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 config.yaml:

middleware:
  cache:
    max_age: 86400
    immutable: true
    public: 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 config.yaml:

middleware:
  cors:
    allow_origins: ["https://example.com", "https://app.example.com"]
    allow_credentials: true
    max_age: 3600

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 after auth middleware. 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.

Exception handling:

Redirect: Returns 3xx redirect with Location header
HTTPException: Returns status code with detail message
Exception: Returns 500 Internal Server Error

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:

middleware:
  errors:
    debug: true  # Show 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 config.yaml:

middleware:
  compression:
    minimum_size: 1000
    compression_level: 6

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 config.yaml:

middleware:
  logging:
    logger_name: "myapp.access"
    level: "DEBUG"
    include_headers: 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