src/genro_asgi/middleware/base.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 base class and the chain mechanism.16 17 ``BaseMiddleware(app, server, **options)`` receives BOTH ends at construction18 (dual parent-child): ``app`` is the next ASGI callable in the chain, ``server``19 the owning server — never discovered by walking wrappers. Subclasses declare20 ``middleware_order`` (lower = outermost; errors 100, logging 200, security 300,21 auth 400, business 500-800, transformation 900) and ``middleware_default``22 (their on/off state when the config does not name them).23 24 ``build_chain(config, innermost, server, registry)`` assembles the chain from25 explicit inputs — ``config`` maps ``{name: bool | dict}`` (a dict value means26 "on" and becomes the middleware's constructor options), ``registry`` maps27 ``{name: class}`` and is always passed in: there is NO module-level registry28 and no import-time registration anywhere. Enabled middlewares are sorted by29 ``middleware_order`` and wrapped innermost-out, so the lowest order ends up30 outermost. A config name missing from the registry raises ``ValueError``.31 32 ``headers_dict(scope)`` parses the ASGI headers into a lowercase-keyed dict33 cached as ``scope["_headers"]`` — the one header-parse shared by the session34 and auth middlewares downstream. ``cookie_value(scope, name)`` reads one35 cookie off it, pair by pair, so a malformed sibling cookie never costs the36 request the cookies that are well-formed.37 """38 39 from __future__ import annotations40 41 import logging42 from http.cookies import CookieError, SimpleCookie43 from typing import TYPE_CHECKING, Any44 45 if TYPE_CHECKING:46 from ..types import ASGIApp, Receive, Scope, Send47 48 __all__ = ["BaseMiddleware", "build_chain", "cookie_value", "headers_dict"]49 50 51 def headers_dict(scope: Scope) -> dict[str, str]:52 """Parse scope headers into a lowercase-keyed dict, cached as ``scope["_headers"]``.53 54 Names and values are decoded as latin-1 per the ASGI spec; duplicate55 headers collapse to the last value.56 """57 headers: dict[str, str] | None = scope.get("_headers")58 if headers is None:59 headers = {60 name.decode("latin-1").lower(): value.decode("latin-1")61 for name, value in scope.get("headers", [])62 }63 scope["_headers"] = headers64 return headers65 66 67 def cookie_value(scope: Scope, name: str) -> str | None:68 """The value of the ``name`` cookie on the request, or ``None``.69 70 Parsed pair by pair: a whole-header ``SimpleCookie.load`` raises71 ``CookieError`` on the first cookie with an illegal key (third-party72 trackers ship them), losing every well-formed sibling with it.73 """74 cookie_header = headers_dict(scope).get("cookie")75 if not cookie_header:76 return None77 jar: SimpleCookie = SimpleCookie()78 for pair in cookie_header.split(";"):79 try:80 jar.load(pair.strip())81 except CookieError:82 continue83 morsel = jar.get(name)84 return morsel.value if morsel is not None else None85 86 87 class BaseMiddleware:88 """Base class for chain middlewares: holds the next app and the server.89 90 Class attributes:91 middleware_order: position in the chain (lower = outermost).92 middleware_default: on/off state when the config does not name it.93 """94 95 middleware_order: int = 50096 middleware_default: bool = False97 98 def __init__(self, app: ASGIApp, server: Any, **options: Any) -> None:99 self._app = app100 self._server = server101 self._logger = logging.getLogger(f"{type(self).__module__}.{type(self).__name__}")102 if options:103 unexpected = ", ".join(sorted(options))104 raise TypeError(105 f"{type(self).__name__}.__init__() got unexpected keyword arguments: {unexpected}"106 )107 108 @property109 def app(self) -> ASGIApp:110 """The next ASGI callable in the chain (towards the base dispatch)."""111 return self._app112 113 @property114 def server(self) -> Any:115 """The owning server, handed in at construction — never walked to."""116 return self._server117 118 @property119 def logger(self) -> logging.Logger:120 """This middleware's instance logger."""121 return self._logger122 123 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:124 """ASGI entry point: concrete middlewares must implement it."""125 raise NotImplementedError(f"{type(self).__name__} does not implement the ASGI callable")126 127 128 def build_chain(129 config: dict[str, bool | dict[str, Any]],130 innermost: ASGIApp,131 server: Any,132 registry: dict[str, type[BaseMiddleware]],133 ) -> ASGIApp:134 """Assemble the middleware chain around ``innermost`` and return its head.135 136 Every middleware named by ``config`` must exist in ``registry``137 (``ValueError`` otherwise); registry entries the config does not name138 follow their ``middleware_default``. A dict config value enables the139 middleware and becomes its constructor options.140 """141 unknown = sorted(name for name in config if name not in registry)142 if unknown:143 raise ValueError(f"unknown middleware name(s) in config: {', '.join(unknown)}")144 enabled: list[tuple[int, type[BaseMiddleware], dict[str, Any]]] = []145 for name, cls in registry.items():146 value = config.get(name)147 if value is None:148 if not cls.middleware_default:149 continue150 options: dict[str, Any] = {}151 elif isinstance(value, dict):152 options = value153 elif value:154 options = {}155 else:156 continue157 enabled.append((cls.middleware_order, cls, options))158 enabled.sort(key=lambda item: item[0])159 app = innermost160 for _order, cls, options in reversed(enabled):161 app = cls(app, server, **options)162 return app