ezmsg.dashboard.backend.app#

Module Attributes

STREAM_STATUS_INTERVAL_SECONDS

How often a stream socket reports tap health and message-inspector state, independently of the data frame rate.

Functions

create_app(graph_service=None, *, frontend_dir=None, execution_supervisor=None)[source]#
Return type:

FastAPI

Parameters:
get_graph_service(request)[source]#
Return type:

GraphServiceProtocol

Parameters:

request (Request)

get_packaged_frontend_dir()[source]#
Return type:

Path | None

async run_until_first_completed(*tasks)[source]#

Await the first task to finish, cancel the rest, re-raise what it raised.

Every websocket route here is the same shape: a task that produces messages and a task that watches for the client going away. Whichever finishes first decides the socket’s fate, and the loser must be cancelled and awaited – leaving it pending would keep the tap it holds alive past the socket that opened it.

Return type:

None

Parameters:

tasks (Task[None])

async wait_for_websocket_disconnect(websocket)[source]#

Return once the peer goes away, ignoring anything it says first.

Return type:

None

Parameters:

websocket (WebSocket)

Classes

class DashboardJSONResponse(content, status_code=200, headers=None, media_type=None, background=None)[source]#

Bases: JSONResponse

JSONResponse that encodes non-finite floats instead of failing on them.

JSONResponse.render renders with allow_nan=False, so a single inf anywhere in a payload turns the whole request into a 500. Payloads assembled by the adapters are already encoded; this is the backstop for everything else (request echoes, future routes). Re-rendering only on failure keeps the common path free of an extra walk over large snapshot payloads.

Parameters:

  • content (Any)

  • status_code (int)

  • headers (Mapping[str, str] | None)

  • media_type (str | None)

  • background (BackgroundTask | None)

render(content)[source]#
Return type:

bytes

Parameters:

content (Any)

class DashboardStaticFiles(*, directory=None, packages=None, html=False, check_dir=True, follow_symlink=False)[source]#

Bases: StaticFiles

Serves the built frontend, and caches it the way Vite builds it.

The shell names the content-hashed bundles, so a cached copy pins the browser to whichever bundle it was built against: a rebuilt dashboard keeps serving the old app until someone hard-reloads. It is therefore never cached. The hashed assets it names can be cached forever by the same logic.

Missing paths that look like client-side routes fall back to the shell.

Parameters:
static cache_headers_for(path)[source]#

Cache headers for a file StaticFiles resolved, if it wants any.

path arrives normalized for the host OS, so it is separated by backslashes on Windows. Rewrite them rather than going through Path, which only reads a backslash as a separator when running on Windows.

Return type:

dict[str, str] | None

Parameters:

path (str)

async get_response(path, scope)[source]#

Returns an HTTP response, given the incoming path, method and request headers.

Return type:

Response

Parameters:
class ExecutionRequest(**data)[source]#

Bases: BaseModel

Parameters:
document: dict[str, Any]#
expected_revision: int#
model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ProfilingTraceControlRequest(**data)[source]#

Bases: BaseModel

Parameters:

  • process_id (str)

  • enabled (bool)

  • publisher_endpoint_id (str | None)

  • publisher_topic (str | None)

  • subscriber_topic (str | None)

  • metrics (list[str] | None)

  • sample_mod (int)

  • ttl_seconds (float | None)

  • timeout (float)

process_id: str#
enabled: bool#
publisher_endpoint_id: str | None#
publisher_topic: str | None#
subscriber_topic: str | None#
metrics: list[str] | None#
sample_mod: int#
ttl_seconds: float | None#
timeout: float#
model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class SettingsFieldPatchRequest(**data)[source]#

Bases: BaseModel

Parameters:
field_path: str#
value: Any#
timeout: float#
model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class StopExecutionRequest(**data)[source]#

Bases: BaseModel

Parameters:

expected_revision (int)

expected_revision: int#
model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].