ezmsg.dashboard.backend.models.events#

Classes

class ProfilingTraceEnvelope(**data)[source]#

Bases: BaseModel

Parameters:
kind: Literal['profiling.trace']#
data: ProfilingTracePayload#
model_config: ClassVar[ConfigDict] = {}#

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

class ProfilingTracePayload(**data)[source]#

Bases: BaseModel

Parameters:
timestamp: float#
batches: dict[str, dict[str, Any]]#
model_config: ClassVar[ConfigDict] = {}#

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

class SettingsChangedEnvelope(**data)[source]#

Bases: BaseModel

Parameters:
kind: Literal['settings.changed']#
data: SettingsChangedPayload#
model_config: ClassVar[ConfigDict] = {}#

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

class SettingsChangedPayload(**data)[source]#

Bases: BaseModel

Parameters:
seq: int#
event_type: str#
component_address: str#
timestamp: float#
source_session_id: str | None#
source_process_id: str | None#
value: SettingsValuePayload#
model_config: ClassVar[ConfigDict] = {}#

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

class SettingsValuePayload(**data)[source]#

Bases: BaseModel

Parameters:
repr_value: dict[str, Any] | str#
structured_value: dict[str, Any] | None#
settings_schema: dict[str, Any] | None#
serialized_present: bool#
model_config: ClassVar[ConfigDict] = {}#

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

class SystemErrorEnvelope(**data)[source]#

Bases: BaseModel

Parameters:
kind: Literal['system.error']#
data: SystemErrorPayload#
model_config: ClassVar[ConfigDict] = {}#

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

class SystemErrorPayload(**data)[source]#

Bases: BaseModel

Parameters:
timestamp: float#
message: str#
model_config: ClassVar[ConfigDict] = {}#

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

class SystemHeartbeatEnvelope(**data)[source]#

Bases: BaseModel

Parameters:
kind: Literal['system.heartbeat']#
data: SystemHeartbeatPayload#
model_config: ClassVar[ConfigDict] = {}#

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

class SystemHeartbeatPayload(**data)[source]#

Bases: BaseModel

Parameters:

timestamp (float)

timestamp: float#
model_config: ClassVar[ConfigDict] = {}#

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

class SystemReadyEnvelope(**data)[source]#

Bases: BaseModel

Parameters:
kind: Literal['system.ready']#
data: SystemReadyPayload#
model_config: ClassVar[ConfigDict] = {}#

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

class SystemReadyPayload(**data)[source]#

Bases: BaseModel

Parameters:
timestamp: float#
message: str#
model_config: ClassVar[ConfigDict] = {}#

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

class TopologyChangedEnvelope(**data)[source]#

Bases: BaseModel

Parameters:
kind: Literal['topology.changed']#
data: TopologyChangedPayload#
model_config: ClassVar[ConfigDict] = {}#

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

class TopologyChangedPayload(**data)[source]#

Bases: BaseModel

Parameters:
  • seq (int)

  • event_type (str)

  • timestamp (float)

  • changed_topics (list[str])

  • source_session_id (str | None)

  • source_process_id (str | None)

seq: int#
event_type: str#
timestamp: float#
changed_topics: list[str]#
source_session_id: str | None#
source_process_id: str | None#
model_config: ClassVar[ConfigDict] = {}#

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