# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""Middleware package - ASGI middleware for genro-asgi."""
from __future__ import annotations
import functools
import importlib
from abc import ABC, abstractmethod
from collections.abc import Callable, Coroutine
from pathlib import Path
from typing import TYPE_CHECKING, Any
from ..utils import decode_headers
if TYPE_CHECKING:
from ..types import ASGIApp, Receive, Scope, Send
MIDDLEWARE_REGISTRY: dict[str, type["BaseMiddleware"]] = {}
_NOT_SET: Any = object()
[docs]
class BaseMiddleware(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"].
"""
middleware_name: str = ""
middleware_order: int = 500
middleware_default: bool = False
__slots__ = ("app", "_cached_server")
[docs]
def __init__(self, app: ASGIApp, **kwargs: Any) -> None:
"""Initialize middleware with wrapped app.
Args:
app: The ASGI app to wrap (next in chain).
**kwargs: Middleware options from the ``root.middleware(name={...})``
entry in the config.py recipe.
"""
self.app = app
self._cached_server: Any = _NOT_SET
@property
def server(self) -> Any:
"""Walk the .app chain to find the AsgiServer via Dispatcher.
Returns None when used standalone (no Dispatcher in chain).
Cached on first access.
"""
if self._cached_server is not _NOT_SET:
return self._cached_server
current = self.app
while hasattr(current, "app"):
current = current.app
result = getattr(current, "server", None)
self._cached_server = result
return result
def __init_subclass__(cls, **kwargs: Any) -> None:
"""Auto-register subclass in MIDDLEWARE_REGISTRY.
Args:
**kwargs: Passed to super().__init_subclass__().
Raises:
ValueError: If middleware_name is already registered.
"""
super().__init_subclass__(**kwargs)
name = cls.middleware_name or cls.__name__
if name in MIDDLEWARE_REGISTRY:
raise ValueError(f"Middleware name '{name}' already registered")
cls.middleware_name = name
MIDDLEWARE_REGISTRY[name] = cls
@abstractmethod
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: ...
def _autodiscover() -> None:
"""Import all middleware modules in this package to trigger registration."""
package_dir = Path(__file__).parent
for py_file in package_dir.glob("*.py"):
if py_file.name.startswith("_"):
continue
module_name = py_file.stem
importlib.import_module(f".{module_name}", __package__)
[docs]
def middleware_chain(
middleware_config: str | list[str] | dict[str, Any],
app: ASGIApp,
) -> ASGIApp:
"""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
Args:
middleware_config: Mapping ``{name: bool | dict}`` (a dict value carries
the middleware options), a comma-separated string of names, or a
list of names.
app: The innermost ASGI app (usually the Dispatcher).
Returns:
Wrapped ASGI app with middleware chain.
"""
# Parse config into {name: value} where value is True/False or an options dict.
config_dict: dict[str, bool | dict[str, Any]] = {}
if isinstance(middleware_config, str):
# "cors, auth" -> all enabled on defaults
for name in middleware_config.split(","):
name = name.strip()
if name:
config_dict[name] = True
elif hasattr(middleware_config, "as_dict"):
# SmartOptions
for name, value in middleware_config.as_dict().items(): # type: ignore[union-attr]
config_dict[name] = value if isinstance(value, dict) else _parse_enabled(value)
elif isinstance(middleware_config, dict):
for name, value in middleware_config.items():
config_dict[name] = value if isinstance(value, dict) else _parse_enabled(value)
elif middleware_config:
# List of names
for name in middleware_config:
config_dict[name] = True
# Collect enabled middleware with their order. A dict value means "on".
enabled: list[tuple[int, str, type[BaseMiddleware]]] = []
for name, cls in MIDDLEWARE_REGISTRY.items():
if name in config_dict:
value = config_dict[name]
# A dict value (even empty) means "on with these options".
is_enabled = True if isinstance(value, dict) else bool(value)
else:
is_enabled = cls.middleware_default
if is_enabled:
enabled.append((cls.middleware_order, name, cls))
# Sort by order (lower first)
enabled.sort(key=lambda x: x[0])
# Build chain (reversed: first in order = outermost wrapper). A dict value
# for a middleware is passed as its constructor kwargs; anything else means
# "on with defaults".
for order, name, cls in reversed(enabled):
value = config_dict.get(name)
config = value if isinstance(value, dict) else {}
app = cls(app, **config)
return app
def _parse_enabled(value: Any) -> bool:
"""Parse on/off/true/false value to bool."""
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.lower() in ("on", "true", "yes", "1")
return bool(value)
_autodiscover()
globals().update(MIDDLEWARE_REGISTRY)
__all__ = [
"BaseMiddleware",
"MIDDLEWARE_REGISTRY",
"headers_dict",
"middleware_chain",
*MIDDLEWARE_REGISTRY.keys(),
]