src/genro_asgi/middleware/__init__.py

Source from this local checkout, regenerated when the reader rebuilds. Line links use #L<number>; a GitHub line range opens its first line.

1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 #     https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Middleware capability: the chain as a mixin over the base server (D16).16 17 The base server has NO middleware. This mixin adds the chain as a capability,18 composed BEFORE the server class (``class MyServer(MiddlewareMixin,19 BaseServer)``): its cooperative ``__init__`` peels ``middleware=`` (the20 ``{name: bool | dict}`` switches; ``None`` arms only the defaults) and21 ``middleware_registry=`` (extra ``{name: class}`` entries merged over22 ``default_registry()``), builds the chain ONCE around ``_base_call`` — the23 adapter delegating to the next ``__call__`` in the MRO, i.e. the base24 dispatch — and routes ONLY ``http`` scopes through it: ``lifespan`` and25 ``websocket`` go straight to ``super().__call__``. A composition WITHOUT the26 mixin simply lacks the attributes — a different type, not a ghost.27 28 ``default_registry()`` returns a FRESH dict per call ({"errors":29 ErrorMiddleware, "wellknown": WellKnownMiddleware, "logging":30 LoggingMiddleware, "cors": CORSMiddleware, "auth": AuthMiddleware, "session":31 SessionMiddleware} as of Phase 5) — deliberately a function so no module-level32 mutable registry exists.33 It lives in this module, not in ``base.py``, because ``base.py`` cannot import34 the concrete middleware modules (which subclass ``BaseMiddleware``) without a35 cycle.36 """37 38 from __future__ import annotations39 40 from typing import TYPE_CHECKING, Any41 42 from .authentication import AuthMiddleware43 from .base import BaseMiddleware, build_chain, headers_dict44 from .cors import CORSMiddleware45 from .errors import ErrorMiddleware46 from .logging import LoggingMiddleware47 from .session import SessionMiddleware48 from .wellknown import WellKnownMiddleware49 50 if TYPE_CHECKING:51     from ..types import ASGIApp, Receive, Scope, Send52 53 __all__ = [54     "AuthMiddleware",55     "BaseMiddleware",56     "CORSMiddleware",57     "ErrorMiddleware",58     "LoggingMiddleware",59     "MiddlewareMixin",60     "SessionMiddleware",61     "WellKnownMiddleware",62     "build_chain",63     "default_registry",64     "headers_dict",65 ]66 67 68 def default_registry() -> dict[str, type[BaseMiddleware]]:69     """A fresh ``{name: class}`` mapping of the middlewares shipped with the core."""70     return {71         "errors": ErrorMiddleware,72         "wellknown": WellKnownMiddleware,73         "logging": LoggingMiddleware,74         "cors": CORSMiddleware,75         "auth": AuthMiddleware,76         "session": SessionMiddleware,77     }78 79 80 class MiddlewareMixin:81     """Middleware capability mixin, composed BEFORE a server class.82 83     Constructor kwargs peeled here: ``middleware`` — the ``{name: bool | dict}``84     switches (a dict value enables the middleware and becomes its constructor85     options); ``middleware_registry`` — extra ``{name: class}`` entries merged86     over ``default_registry()``.87     """88 89     def __init__(self, **kwargs: Any) -> None:90         middleware: dict[str, bool | dict[str, Any]] | None = kwargs.pop("middleware", None)91         extra: dict[str, type[BaseMiddleware]] | None = kwargs.pop("middleware_registry", None)92         super().__init__(**kwargs)93         registry = default_registry()94         if extra:95             registry.update(extra)96         self._middleware_chain = build_chain(middleware or {}, self._base_call, self, registry)97 98     @property99     def middleware_chain(self) -> ASGIApp:100         """The assembled chain: outermost middleware first, base dispatch innermost."""101         return self._middleware_chain102 103     def get_middleware(self, middleware_class: type[BaseMiddleware]) -> BaseMiddleware | None:104         """The layer of that class in this chain, or ``None`` when it is off.105 106         Args:107             middleware_class: the class to look for; a subclass of it answers too.108 109         Returns:110             The assembled instance, or ``None`` — a middleware nobody enabled111             has none.112 113         The chain is walked from its head: every layer holds the next one, so114         the instances need not be kept a second time. What reads this is code115         that must do for a websocket what a middleware does for HTTP — the116         handshake asking ``SessionMiddleware`` for the session of a scope the117         chain never saw.118         """119         layer = self.middleware_chain120         while isinstance(layer, BaseMiddleware):121             if isinstance(layer, middleware_class):122                 return layer123             layer = layer.app124         return None125 126     async def _base_call(self, scope: Scope, receive: Receive, send: Send) -> None:127         """Innermost chain target: the next ``__call__`` after this mixin in the MRO."""128         await super().__call__(scope, receive, send)129 130     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:131         """Route ``http`` scopes through the chain; every other scope passes straight through."""132         if scope["type"] != "http":133             await super().__call__(scope, receive, send)134             return135         await self.middleware_chain(scope, receive, send)