exlab_wizard.api.app#

FastAPI app + lifespan + dependency wiring. Backend Spec §4.6.

The create_app() factory builds the app with the §4.6 versioned prefix (/api/v1/...), mounts every router, registers exception handlers (§4.6.3 envelope), and binds an AppDependencies instance onto app.state.dependencies so per-request handlers can look up the live controller / validator / cache writers / etc.

Lifespan responsibilities (§4.5):

  • Load (or accept the supplied) config.yaml.

  • Build the plugin registry (best-effort; failure logs WARN).

  • Refresh the LIMS project cache (best-effort).

  • Start the background audit task (every 30 s; pub-sub publishes deltas on the /problems/events channel).

  • On shutdown, drain in-flight sessions, stop the audit task, and close the cache writers.

The launcher (in production) constructs a full AppDependencies with real components; tests can pass a custom dependencies object whose fields are stubs / mocks.

Functions

create_app(*[, config, dependencies, ...])

Build the FastAPI app.

Classes

AppDependencies([config, save_config, ...])

Bundle of live components the API surface dispatches to.

AuditChannel()

Multi-subscriber pub-sub for the Problems WebSocket.
class exlab_wizard.api.app.AppDependencies(config=None, save_config=None, lims_reachable=True, keyring_password_present=True, lims_reason=None, controller=None, validator=None, plugin_host=None, cache_creation=None, lims_client=None, nas_sync=None, session_store=None, ingest_writer=None, staging_watcher=None, audit_channel=None, last_audit_at=None, nas_sync_snapshot=None, session_store_snapshot=None, registered_plugin_count=0, plugin_host_status='ok', lims_probe=None, equipment_probe=None, autostart_toggle=None, audit_task=None)[source]#

Bases: object

Bundle of live components the API surface dispatches to.

Production wiring (the launcher) constructs everything; tests can pass mocks. Attributes are typed loosely (Any) so the API code does not impose imports on the caller – the runtime contract is documented per attribute.

Parameters:
audit_channel: AuditChannel | None = None#
audit_task: Task[None] | None = None#
autostart_toggle: Callable[[bool], Any] | None = None#
cache_creation: Any = None#
config: Config | None = None#
controller: Any = None#
equipment_probe: Callable[[...], Any] | None = None#
ingest_writer: Any = None#
keyring_password_present: bool = True#
last_audit_at: str | None = None#
lims_client: Any = None#
lims_probe: Callable[[...], Any] | None = None#
lims_reachable: bool = True#
lims_reason: str | None = None#
nas_sync: Any = None#
nas_sync_snapshot: Callable[[], dict[str, Any]] | None = None#
plugin_host: Any = None#
plugin_host_status: str = 'ok'#
registered_plugin_count: int = 0#
save_config: Callable[[Config], Awaitable[None] | None] | None = None#
session_store: Any = None#
session_store_snapshot: Callable[[], dict[str, Any]] | None = None#
staging_watcher: Any = None#
validator: Any = None#
class exlab_wizard.api.app.AuditChannel[source]#

Bases: object

Multi-subscriber pub-sub for the Problems WebSocket. Backend Spec §4.6.2.

Subscribers receive every published frame (snapshot or delta). The channel keeps the most recent snapshot so late subscribers do not have to wait for the next 30-second tick.

close()[source]#

Close every subscriber’s queue. Idempotent.

Return type:

None

async publish_delta(*, added, removed, changed, audit_at)[source]#
Parameters:
Return type:

None

async publish_snapshot(findings, audit_at)[source]#
Parameters:
Return type:

None

subscribe()[source]#

Return an async iterator that yields every published frame.

Return type:

AsyncIterator[dict[str, Any]]

exlab_wizard.api.app.create_app(*, config=None, dependencies=None, audit_interval_seconds=30.0, start_audit_task=False)[source]#

Build the FastAPI app. Backend Spec §4.6.

config: optional pre-loaded config.yaml; if dependencies is supplied this is ignored. dependencies: a fully-configured AppDependencies (production launcher uses this). audit_interval_seconds: how often the background audit task runs; tests can pass a small value to exercise the loop. start_audit_task: if True the lifespan handler launches the audit task; defaults to False so tests don’t accumulate tasks.

Parameters:
Return type:

FastAPI