src/genro_asgi/request_registry.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 """Request registry: the current request and the in-flight picture.16 17 ``RequestRegistry`` is held by the server as a dual parent-child18 (``self.server``, SPECIFICATION.md §4) and is the SINGLE writer of the19 in-flight set (D12 spirit): the server registers a request on entry and20 unregisters it on exit, around the http dispatch. Each registration is a21 lightweight ``RegisteredRequest`` record (D18: slotted, high cardinality)22 carrying a monotonic id, the scope type, the path, the start time, and the23 request's cleanup callbacks. The server drains those cleanups in the http24 ``finally`` (``run_cleanups``) so any app — routed or bare — gets end-of-request25 teardown (e.g. ``request.db`` closing its connection) for free.26 27 The "current request" is exposed through a ContextVar that lives on the28 registry INSTANCE (never at module level): ``register`` sets it and keeps the29 reset token, ``unregister`` resets it. Because the ContextVar is an instance30 attribute, deleting the server garbage-collects everything (instance-isolation31 rule) and concurrent requests — each on its own task context — see their own32 ``current``.33 """34 35 from __future__ import annotations36 37 import asyncio38 import contextlib39 import logging40 import time41 from collections.abc import Callable42 from contextvars import ContextVar, Token43 from typing import TYPE_CHECKING, Any44 45 if TYPE_CHECKING:46 from .server import BaseServer47 from .types import Scope48 49 __all__ = ["RegisteredRequest", "RequestRegistry"]50 51 52 class RegisteredRequest:53 """One in-flight request tracked by the registry (D18: slotted record).54 55 A snapshot taken at registration: the monotonic ``request_id``, the ASGI56 ``scope_type``, the ``path``, and ``started_at`` (``time.monotonic()``).57 Slotted because requests are high cardinality.58 59 It also owns the request's end-of-life cleanups: ``add_cleanup(fn)`` queues60 a zero-arg callback and ``run_cleanups()`` drains them LIFO at the end of the61 dispatch (the server calls it in the http ``finally``). The ``_cleanups``62 list is lazy — allocated only when the first callback is queued — so a63 request that registers none pays nothing.64 """65 66 __slots__ = ("_request_id", "_scope_type", "_path", "_started_at", "_cleanups")67 68 def __init__(self, request_id: int, scope_type: str, path: str) -> None:69 self._request_id = request_id70 self._scope_type = scope_type71 self._path = path72 self._started_at = time.monotonic()73 self._cleanups: list[Callable[[], Any]] | None = None74 75 @property76 def request_id(self) -> int:77 """Monotonic id assigned by the registry at registration."""78 return self._request_id79 80 @property81 def scope_type(self) -> str:82 """ASGI scope type of the request (``http``)."""83 return self._scope_type84 85 @property86 def path(self) -> str:87 """Request path at registration time."""88 return self._path89 90 @property91 def started_at(self) -> float:92 """``time.monotonic()`` captured at registration."""93 return self._started_at94 95 def add_cleanup(self, callback: Callable[[], Any]) -> None:96 """Queue a zero-arg ``callback`` to run at end of request (LIFO)."""97 if self._cleanups is None:98 self._cleanups = []99 self._cleanups.append(callback)100 101 def run_cleanups(self, error: BaseException | None = None) -> None:102 """Run queued cleanups LIFO, isolating and logging each one's exception.103 104 Called by the server in the http ``finally`` — so cleanups run whether105 the request succeeded or failed. ``error`` carries the terminating106 exception (``None`` on success) for error-aware cleanups; the base drain107 runs every callback regardless.108 """109 if self._cleanups is None:110 return111 for callback in reversed(self._cleanups):112 try:113 callback()114 except Exception:115 logging.getLogger(__name__).exception("Request cleanup %r failed", callback)116 117 def __repr__(self) -> str:118 return (119 f"<RegisteredRequest id={self.request_id} "120 f"type={self.scope_type} path={self.path!r}>"121 )122 123 124 class RequestRegistry:125 """Tracks in-flight requests and the current one, owned by the server.126 127 The server is the single writer: it calls ``register(scope)`` on request128 entry and ``unregister(item)`` on exit. ``current`` reads the instance-owned129 ContextVar; ``in_flight`` counts the live requests; ``snapshot()`` lists them.130 """131 132 def __init__(self, server: BaseServer) -> None:133 self.server = server134 self._counter = 0135 self._in_flight: dict[int, RegisteredRequest] = {}136 self._tokens: dict[int, Token[RegisteredRequest | None]] = {}137 self._current: ContextVar[RegisteredRequest | None] = ContextVar(138 "current_request", default=None139 )140 self._empty = asyncio.Event()141 self._empty.set()142 143 @property144 def current(self) -> RegisteredRequest | None:145 """The request being handled in this task's context, or ``None``."""146 return self._current.get()147 148 @property149 def in_flight(self) -> int:150 """How many requests are registered right now."""151 return len(self._in_flight)152 153 def register(self, scope: Scope) -> RegisteredRequest:154 """Register a request from ``scope``, set ``current``, return the item."""155 self._counter += 1156 item = RegisteredRequest(self._counter, scope["type"], scope["path"])157 self._in_flight[item.request_id] = item158 self._tokens[item.request_id] = self._current.set(item)159 self._empty.clear()160 return item161 162 def unregister(self, item: RegisteredRequest) -> None:163 """Drop ``item`` from the in-flight set and reset ``current``."""164 self._in_flight.pop(item.request_id)165 self._current.reset(self._tokens.pop(item.request_id))166 if not self._in_flight:167 self._empty.set()168 169 async def await_drain(self, timeout: float | None = None) -> int:170 """Wait for the in-flight set to empty.171 172 Args:173 timeout: seconds to wait, or None to wait without a bound.174 175 Returns:176 How many requests are still in flight — zero when the drain177 completed, the count to report when the timeout ran out first.178 """179 with contextlib.suppress(TimeoutError):180 await asyncio.wait_for(self._empty.wait(), timeout)181 return self.in_flight182 183 def snapshot(self) -> list[RegisteredRequest]:184 """A list of the currently in-flight requests (registration order)."""185 return list(self._in_flight.values())