exlab_wizard.api#

HTTP API package. Backend Spec §4.

Public re-exports so callers (the launcher, tests) can import the factory and dependency types without reaching into submodules.

class exlab_wizard.api.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.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.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

Modules

app

FastAPI app + lifespan + dependency wiring.

errors

Error envelope helpers + FastAPI exception handlers.

events

WebSocket frame envelope types.

health

GET /api/v1/health rollup.

routers

API routers package.

schemas

msgspec.Struct types for the ExLab-Wizard cache files.

setup

Setup-state gate + /setup/* endpoints.