src/genro_asgi/server.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 """The base server: the applications it serves, ASGI dispatch, uvicorn boot.16 17 ``BaseServer`` is the common substrate of every server (SPECIFICATION.md §4,18 D2): it is composed with its applications (``applications=`` kwarg, a list)19 and keeps them in a dict keyed by each app's ``code``, plus a private index by20 ``mount`` — the one demux mechanism of D3. The set of applications is fixed at21 construction; registration is internal. At the base,22 ``authenticate()`` answers nobody (``None``) and ``session()`` answers none23 (``None``). It owns exactly one thread pool (D2): ``run_sync()`` dispatches a24 blocking handler onto it via ``loop.run_in_executor`` while async handlers stay25 on the loop; the pool is provisioned lazily on first use and torn down at26 shutdown.27 28 As an ASGI callable, ``__call__`` dispatches on the scope type: ``http`` runs29 the D3 demux — first path segment → the app mounted there with that segment30 stripped; else the app on the site root with the full path; else a 307 from31 ``/`` to the declared ``default``; else 404 — ``websocket`` runs32 ``on_websocket``,33 whose DEFAULT is the empty socket of D7 (accepts nothing, closes cleanly with34 code 1000); ``lifespan`` runs the ``Lifespan`` handler (ordered startup,35 reverse shutdown, error isolation). Each http dispatch is registered in the36 ``RequestRegistry`` (``requests``) for the span of the request — the current37 request and the in-flight picture. ``serve()`` boots uvicorn programmatically.38 39 Cooperative init (D16): peels its own kwargs (``applications``,40 ``max_threads``) and, as the end of the chain, raises ``TypeError`` naming any41 leftover kwargs. Mixins go BEFORE ``BaseServer`` in the MRO.42 43 Ownership channel (one direction): registering an application assigns44 ``app.server = self``; the app-side setter enforces exactly-once.45 """46 47 from __future__ import annotations48 49 from typing import TYPE_CHECKING, Any, Callable, Iterable50 51 import uvicorn52 53 from .application import BaseApplication54 from .lifespan import QUITTING, RUNNING, STOPPING, Lifespan55 from .pool import WorkPool56 from .request_registry import RequestRegistry57 from .response import Response58 from .websocket import WebSocket, WebSocketRegistry59 from .wsx_payload import SerializedWsxPayload60 from .wsx import WsxConnection, WsxEnvelope61 62 if TYPE_CHECKING:63 from .types import ASGIApp, Receive, Scope, Send64 65 REFUSED_RETRY_AFTER_SECONDS = 566 """The seconds a refused request is told to come back in."""67 68 WEBSOCKET_MAX_CONCURRENT = 1669 #: How long uvicorn waits for open connections at shutdown before cancelling them.70 SHUTDOWN_TIMEOUT_SECONDS = 5.071 """How many messages of ONE websocket connection may be served at once.72 73 A setpoint (owner, 2026-09-06: «configurabile default 16»): the ceiling is what74 keeps a client that floods from sinking the server.75 """76 77 __all__ = [78 "QUITTING",79 "REFUSED_RETRY_AFTER_SECONDS",80 "RUNNING",81 "STOPPING",82 "WEBSOCKET_MAX_CONCURRENT",83 "BaseServer",84 ]85 86 87 class BaseServer:88 """Base server owning the applications it was composed with.89 90 Constructor kwargs peeled here: ``applications`` — the applications this91 server serves — ``default`` — the ``code`` of the application ``/``92 redirects to when nothing answers the root (an unknown code raises93 ``ValueError``) — ``max_threads`` — the pool's worker count, handed to94 ``WorkPool`` (``None`` keeps the stdlib default) — ``websocket`` — the95 websocket options, ``{"origins": [...], "max_concurrent": 16}`` — and96 ``shutdown_timeout_seconds`` — how long uvicorn waits for open connections97 to finish before it cancels them at shutdown (5.0). Without a bound, one98 endless response — an SSE stream a client never closes — holds the server99 for ever and the lifespan shutdown never runs (measured 2026-09-08).100 """101 102 def __init__(self, **kwargs: Any) -> None:103 applications: Iterable[BaseApplication] = kwargs.pop("applications", ())104 default: str | None = kwargs.pop("default", None)105 max_threads: int | None = kwargs.pop("max_threads", None)106 debug: bool | str = kwargs.pop("debug", False)107 websocket: dict[str, Any] = kwargs.pop("websocket", None) or {}108 shutdown_timeout: float = float(109 kwargs.pop("shutdown_timeout_seconds", None) or SHUTDOWN_TIMEOUT_SECONDS110 )111 if kwargs:112 unexpected = ", ".join(sorted(kwargs))113 raise TypeError(114 f"{type(self).__name__}.__init__() got unexpected keyword arguments: {unexpected}"115 )116 super().__init__()117 self._applications: dict[str, BaseApplication] = {}118 self._by_mount: dict[str, BaseApplication] = {}119 self._databases: dict[str, Any] = {}120 self._uvicorn: uvicorn.Server | None = None121 self._pool = WorkPool(self, max_threads=max_threads)122 self._lifespan = Lifespan(self)123 self._registry = RequestRegistry(self)124 self._websockets = WebSocketRegistry()125 self._websocket_origins: list[str] = list(websocket.get("origins") or [])126 self._websocket_max_concurrent: int = int(127 websocket.get("max_concurrent") or WEBSOCKET_MAX_CONCURRENT128 )129 self._shutdown_timeout_seconds = shutdown_timeout130 self.state = RUNNING131 """``RUNNING``, ``QUITTING`` or ``STOPPING`` — read by the entry point."""132 self.shutdown_mode = STOPPING133 """What ``state`` becomes at the lifespan shutdown when nobody chose first.134 135 ``STOPPING`` — down dry — unless the trigger declares its exit saves:136 the ``--reload`` launcher sets ``QUITTING`` here, the deliberate command137 will set ``state`` itself before the shutdown arrives.138 """139 self.debug = debug140 """The declared usage mode: False, True, or the parameters it was given.141 142 A flag and nothing else (owner, 2026-08-25): the core branches on it143 nowhere. It exists so future readers — extra middleware, extra checks —144 can behave differently knowing the server runs in debug.145 """146 for app in applications:147 self.register_application(app)148 self._default = default149 if default is not None and default not in self.applications:150 raise ValueError(f"default names no served application: {default!r}")151 152 @property153 def applications(self) -> dict[str, BaseApplication]:154 """The served applications keyed by their ``code``."""155 return self._applications156 157 @property158 def root_application(self) -> BaseApplication | None:159 """The application on the site root (``mount == ""``), ``None`` if there is none.160 161 It answers ``/`` and every path no other mount claims. A server of162 mounts only has none: then ``/`` redirects to the ``default`` if one is163 declared, and an unclaimed path is a 404.164 """165 return self.application_at("")166 167 @property168 def default_application(self) -> BaseApplication | None:169 """The application ``/`` redirects to, ``None`` if no ``default`` was declared.170 171 It elects nothing: the redirect is the whole of its meaning, and it is172 only consulted when no application answers the root.173 """174 return self.applications[self._default] if self._default is not None else None175 176 def application_at(self, mount: str) -> BaseApplication | None:177 """The application answering under the URL prefix ``mount`` (``None`` if none)."""178 return self._by_mount.get(mount)179 180 def register_application(self, app: BaseApplication) -> None:181 """Register ``app`` under its ``code`` and its ``mount``.182 183 Assigns the ownership channel (``app.server = self``). Internal: the184 set of applications is fixed at construction, so the callers are185 ``__init__`` and the composition layers building a server. A claimed186 code and a claimed mount both raise ``ValueError``.187 """188 mount = app.code if app.mount is None else app.mount189 if app.code in self.applications:190 raise ValueError(f"application code already claimed: {app.code}")191 if mount in self._by_mount:192 raise ValueError(f"mount already claimed: {mount!r}")193 app.server = self194 self.applications[app.code] = app195 self._by_mount[mount] = app196 197 @property198 def databases(self) -> dict[str, Any]:199 """Database handlers keyed by their config ``code`` (may be empty)."""200 return self._databases201 202 def add_database(self, code: str, handler: Any) -> None:203 """Register ``handler`` under ``code``. A claimed code raises ``ValueError``."""204 if code in self.databases:205 raise ValueError(f"database code already registered: {code}")206 self.databases[code] = handler207 208 @property209 def lifespan(self) -> Lifespan:210 """The ``Lifespan`` handler managing this server's startup/shutdown."""211 return self._lifespan212 213 @property214 def pool(self) -> WorkPool:215 """The server's single thread pool for blocking (sync) handlers."""216 return self._pool217 218 @property219 def requests(self) -> RequestRegistry:220 """The registry of in-flight requests and the current one."""221 return self._registry222 223 async def run_sync(self, fn: Callable[..., Any], *args: Any) -> Any:224 """Dispatch blocking ``fn`` onto the pool (the app-side sync protocol).225 226 Apps call ``self.server.run_sync(...)`` for blocking work so it runs227 off the event loop; async handlers simply stay on the loop and never228 touch the pool.229 """230 return await self.pool.run(fn, *args)231 232 def authenticate(self, request: Any) -> Any:233 """Base answer: nobody (``None``). Auth capabilities override this."""234 return None235 236 def session(self, request: Any) -> Any:237 """Base answer: none (``None``). Session capabilities override this."""238 return None239 240 def get_middleware(self, middleware_class: type) -> Any:241 """Base answer: none (``None``). The middleware capability overrides this."""242 return None243 244 @property245 def websockets(self) -> WebSocketRegistry:246 """The live websockets, and which one each page speaks on."""247 return self._websockets248 249 async def send_message(self, page_id: str, path: str, data: Any = None) -> bool:250 """Write one message of the server's own onto the socket a page speaks on.251 252 Args:253 page_id: the page to address.254 path: what the client routes the message on, the way this server255 routes what the client sends.256 data: the payload, as a Python value.257 258 Returns:259 ``True`` when the message was written to a socket, ``False`` when260 that page speaks on none or its socket is already closed.261 262 The message has the shape of a request and carries NO ``id``: it is not263 an answer, and nobody answers it — a page that wants to reply sends an264 rpc of its own, on the ``reply_path`` it asked for or on a path of its265 choosing. DELIVERED means written to the socket, never executed by the266 page: nothing here waits for anything.267 """268 socket = self.websockets.get_page_socket(page_id)269 if socket is None or not socket.connected:270 return False271 envelope = WsxEnvelope(method="WSK", path=path, data=data, page_id=page_id)272 await socket.send_text(envelope.encode())273 return True274 275 async def send_serialized_message(276 self, page_id: str, path: str, payload: SerializedWsxPayload277 ) -> bool:278 """Forward an explicitly serialized application value to its page."""279 socket = self.websockets.get_page_socket(page_id)280 if socket is None or not socket.connected:281 return False282 await socket.send_text(WsxEnvelope(method="WSK", path=path,283 serialized_data=payload, page_id=page_id).encode())284 return True285 286 @property287 def websocket_origins(self) -> list[str]:288 """The Origins a handshake may come from; empty means same-origin only."""289 return self._websocket_origins290 291 @property292 def websocket_max_concurrent(self) -> int:293 """How many messages of ONE connection may be in flight at once."""294 return self._websocket_max_concurrent295 296 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:297 """ASGI entry point: dispatch on the scope type.298 299 ``http`` runs the D3 demux (registering the request in ``requests`` for300 the span of the dispatch); ``websocket`` runs ``on_websocket`` (the301 empty socket by default); ``lifespan`` runs the ``Lifespan`` handler.302 Any other type is an ASGI protocol error.303 304 ``state`` is read FIRST: anything but ``RUNNING`` takes nothing new in305 charge, and this branch renders the refusal the way HTTP says it — 503306 and ``Retry-After``. Another transport reads the same state and renders307 its own. Whatever the middleware chain answers by itself never reaches308 here, so it is neither registered nor refused.309 """310 scope_type = scope["type"]311 if scope_type == "http":312 if self.state != RUNNING:313 await Response(314 content="Server restarting",315 status_code=503,316 media_type="text/plain",317 headers={"retry-after": str(REFUSED_RETRY_AFTER_SECONDS)},318 )(scope, receive, send)319 return320 item = self.requests.register(scope)321 try:322 app, target = self.demux(scope)323 await app(target, receive, send)324 finally:325 item.run_cleanups()326 self.requests.unregister(item)327 elif scope_type == "websocket":328 await self.on_websocket(scope, receive, send)329 elif scope_type == "lifespan":330 await self.lifespan(scope, receive, send)331 # The lifespan handler returns once shutdown is acked; tear the332 # pool down here (a no-op unless a sync dispatch provisioned it).333 self.pool.shutdown(wait=True)334 else:335 raise ValueError(f"unsupported ASGI scope type: {scope_type}")336 337 def demux(self, scope: Scope) -> tuple[ASGIApp, Scope]:338 """D3 demux: pick what answers an http scope, and the scope it receives.339 340 One rule, four branches: the first path segment matching a mount → that341 app, with the segment stripped from ``path`` (the forwarded path is342 rebuilt from the same remainder used to find the segment, so ``//api/x``343 forwards ``/x``); else the application on the site root, with the full344 path unchanged; else, for ``/`` itself with a ``default`` declared, a345 **307** to that application's mount carrying the query string over;346 else **404**. ``/`` on a server WITH a root application matches its347 empty mount in the first branch, which forwards the same ``/``.348 """349 path = scope["path"]350 rest = path.lstrip("/")351 segment, _, remainder = rest.partition("/")352 app = self.application_at(segment)353 if app is not None:354 sub_scope = dict(scope)355 sub_scope["path"] = "/" + remainder356 return app, sub_scope357 root = self.root_application358 if root is not None:359 return root, scope360 default = self.default_application if not rest else None361 if default is not None:362 return self.redirect_to_default(default, scope), scope363 return Response(content="Not Found", status_code=404, media_type="text/plain"), scope364 365 def redirect_to_default(self, app: BaseApplication, scope: Scope) -> Response:366 """A 307 to ``app``'s mount, preserving the query string.367 368 307 and not 301/302: the method and the body must survive the hop, so a369 ``POST /`` reaches the default application as a POST.370 """371 location = f"/{app.mount}/"372 query = scope.get("query_string", b"")373 if query:374 location = f"{location}?{query.decode('latin-1')}"375 return Response(status_code=307, headers={"location": location})376 377 async def on_websocket(self, scope: Scope, receive: Receive, send: Send) -> None:378 """Live one websocket connection: the motor, or the application's own hands.379 380 One ``WsxConnection`` per socket does the whole thing (#68): it judges381 the handshake, accepts it, turns every message into a request the demux382 routes like any other, and answers the ones that carry an ``id``. The383 connection is registered in ``websockets`` for its whole life.384 385 The state is judged FIRST, above the demux, for every websocket: a386 server that is not ``RUNNING`` takes no new connection in charge, and387 the handshake is turned away before the accept. The browser sees the388 handshake fail with no readable code — 1013 exists only after an accept,389 and in the raw mode the accept belongs to the application — but the390 state is the machine's business and not the protocol's, exactly as it391 is on the http branch.392 393 The exception is an application that wants the socket ITSELF: the394 handshake's path names it through the same demux, and if it defines395 ``serve_websocket`` it is handed the raw scope, receive and send, with396 the segment of its mount already taken off the path. Nothing else of the397 motor runs then — no accept, no Origin gate, no registry: an application398 that takes the socket takes all of it, and the core does not half-serve399 a connection it does not hold. It is the admitted mode of the design,400 the one a hosted framework with a websocket protocol of its own reaches401 the server by.402 """403 if self.state != RUNNING:404 await WebSocket(scope, receive, send).refuse(1013, "server restarting")405 return406 app, target = self.demux(scope)407 raw_seam = getattr(app, "serve_websocket", None)408 if raw_seam is not None:409 await raw_seam(target, receive, send)410 return411 await WsxConnection(self, scope, receive, send).serve()412 413 @property414 def shutdown_timeout_seconds(self) -> float:415 """How long uvicorn waits for open connections at shutdown before cancelling them."""416 return self._shutdown_timeout_seconds417 418 @property419 def uvicorn_server(self) -> uvicorn.Server | None:420 """The uvicorn ``Server`` once ``serve()`` has built it (else ``None``).421 422 Callers that boot the server in a background thread read the bound port423 from ``uvicorn_server.servers[0].sockets[0].getsockname()`` after424 ``uvicorn_server.started`` turns true.425 """426 return self._uvicorn427 428 def serve(self, host: str = "127.0.0.1", port: int = 0) -> None:429 """Boot uvicorn programmatically, serving this server (blocking).430 431 Builds ``uvicorn.Config``/``uvicorn.Server`` and runs it. ``port=0``432 lets the OS assign an ephemeral port, discoverable via433 ``uvicorn_server`` once started. ``shutdown_timeout_seconds`` bounds434 uvicorn's wait for open connections, so a response that never ends435 cannot keep the lifespan shutdown — and the applications' own stop —436 from running.437 """438 self._uvicorn = uvicorn.Server(439 uvicorn.Config(440 self,441 host=host,442 port=port,443 timeout_graceful_shutdown=self.shutdown_timeout_seconds,444 )445 )446 self._uvicorn.run()