ezmsg.dashboard.backend.services.graph_context_service#

Classes

class GraphContextLifecycleService(*, graph_address=None, auto_start=None, heartbeat_seconds=10.0, queue_size=512, graph_context_factory=<class 'ezmsg.core.graphcontext.GraphContext'>)[source]#

Bases: object

Parameters:

graph_address (Any)
auto_start (bool | None)
heartbeat_seconds (float)
queue_size (int)
graph_context_factory (Callable[..., GraphContext])

__init__(*, graph_address=None, auto_start=None, heartbeat_seconds=10.0, queue_size=512, graph_context_factory=<class 'ezmsg.core.graphcontext.GraphContext'>)[source]#

Parameters:

graph_address (Any)
auto_start (bool | None)
heartbeat_seconds (float)
queue_size (int)
graph_context_factory (Callable[[...], GraphContext])

Return type:

None

property is_started: bool#

async startup()[source]#

Return type:: None

async shutdown()[source]#

Return type:: None

async health_payload()[source]#

Return type:: dict[str, Any]

stream_tap(*, topic, mode, max_columns, window_seconds=2.0)[source]#

Open (or join) a tap on topic for the duration of the context.

Not async def: callers use it as async with, and returning the context manager directly means a caller that never enters it has not already taken a reference on the tap.

Return type:

AbstractAsyncContextManager[StreamTapClient]

Parameters:

topic (str)
mode (str)
max_columns (int)
window_seconds (float)

async snapshot_payload()[source]#

Return type:: dict[str, Any]

async settings_payload()[source]#

Return type:: dict[str, Any]

async update_setting_field(*, component_address, field_path, value, timeout=2.0)[source]#

Return type:

dict[str, Any]

Parameters:

component_address (str)
field_path (str)
value (Any)
timeout (float)

async set_profiling_trace_control(*, process_id, enabled, publisher_endpoint_id, publisher_topic, subscriber_topic, metrics, sample_mod=1, ttl_seconds=None, timeout=2.0)[source]#

Return type:

dict[str, Any]

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)

async event_envelopes(*, topology_after_seq=0, settings_after_seq=0, profiling_interval=0.05, profiling_max_samples=5000)[source]#

Return type:

AsyncIterator[dict[str, Any]]

Parameters:

topology_after_seq (int)
settings_after_seq (int)
profiling_interval (float)
profiling_max_samples (int)

class GraphServiceProtocol(*args, **kwargs)[source]#

Bases: Protocol

async startup()[source]#

Return type:: None

async shutdown()[source]#

Return type:: None

async health_payload()[source]#

Return type:: dict[str, Any]

async snapshot_payload()[source]#

Return type:: dict[str, Any]

async settings_payload()[source]#

Return type:: dict[str, Any]

async update_setting_field(*, component_address, field_path, value, timeout)[source]#

Return type:

dict[str, Any]

Parameters:

component_address (str)
field_path (str)
value (Any)
timeout (float)

async set_profiling_trace_control(*, process_id, enabled, publisher_endpoint_id, publisher_topic, subscriber_topic, metrics, sample_mod, ttl_seconds, timeout)[source]#