Skip to main content

server

Runs the patient-data API via uvicorn in a dedicated daemon thread.

Mirrors the threading model of _PodControlServer: the ASGI server runs on its own event loop in a separate thread so it does not interfere with the pod's main event loop.

Classes​

PatientAPIServer​

class PatientAPIServer(    repository: PatientDataRepository,    port: int,    verifier: JWTVerifier,    host: str | None = None,    ssl_certfile: Path | None = None,    ssl_keyfile: Path | None = None,    allowed_origins: list[str] | None = None,):

Hosts the patient-data API in a background daemon thread.

Arguments

  • repository: The data-access implementation to serve.
  • port: The TCP port to listen on.
  • verifier: The JWT verifier used to authenticate requests.
  • host: The interface to bind to. Defaults to loopback, unless ssl_certfile/ssl_keyfile are provided, in which case it defaults to 0.0.0.0 so that it's available from the network.
  • ssl_certfile: Path to a PEM file containing the leaf certificate (optionally followed by the CA bundle) to serve over HTTPS. Must be provided together with ssl_keyfile.
  • ssl_keyfile: Path to the PEM private key matching ssl_certfile.
  • allowed_origins: Browser origins permitted to read the API cross-origin via CORS. Empty by default (no CORS headers).

Initialise the server with the repository, port, verifier, and host.

Methods​

start​

def start(self, thread_name: str = 'patient_api_server', daemon: bool = True) ‑> bool:

Start serving in a background thread.

uvicorn.Server.run() binds the socket (and, for HTTPS, loads the certificate/key) as the first step of its own async setup, inside the background thread; only once that has succeeded does it flip Server.started to True. A failure there (port still held by a not-yet-exited previous instance, unreadable/invalid cert or key) doesn't raise back into this method - uvicorn either calls sys.exit(1), which is swallowed at the thread boundary, or the thread just dies with an uncaught exception - so the only way for the caller to distinguish "started" from "failed" is to wait for Server.started or for the thread to die. Callers rely on this to know whether it's safe to treat the server as actually serving (see Pod._run_patient_api_server, which only records the certificate as picked up once this returns True).

Arguments

  • thread_name: Name for the server thread.
  • daemon: Whether the thread is a daemon thread.

Returns True if startup was confirmed within _STARTUP_CONFIRM_TIMEOUT seconds, False otherwise.

stop​

def stop(self) ‑> None:

Signal graceful shutdown and wait for the thread to exit.