Skip to content

Telemetry

Textual-free telemetry source and metric contracts.

MetricValue dataclass 

MetricValue(
    key: str,
    label: str,
    value: float | int | str | None,
    unit: str | None,
    status: SemanticStatus,
    source_id: str,
    scope: str | None = None,
)

key instance-attribute 

key: str

label instance-attribute 

label: str

scope class-attribute instance-attribute 

scope: str | None = None

source_id instance-attribute 

source_id: str

status instance-attribute 

status: SemanticStatus

unit instance-attribute 

unit: str | None

value instance-attribute 

value: float | int | str | None

SourceAvailability dataclass 

SourceAvailability(
    source_id: str,
    status: SourceStatus,
    capabilities: frozenset[str] = frozenset(),
    message: str | None = None,
)

available property 

available: bool

capabilities class-attribute instance-attribute 

capabilities: frozenset[str] = frozenset()

message class-attribute instance-attribute 

message: str | None = None

source_id instance-attribute 

source_id: str

status instance-attribute 

status: SourceStatus

SourceStatus

Bases: StrEnum

AVAILABLE class-attribute instance-attribute 

AVAILABLE = 'available'

DEGRADED class-attribute instance-attribute 

DEGRADED = 'degraded'

UNAVAILABLE class-attribute instance-attribute 

UNAVAILABLE = 'unavailable'

TelemetrySnapshot dataclass 

TelemetrySnapshot(
    source_id: str,
    sampled_at: datetime,
    capabilities: frozenset[str],
    metrics: tuple[MetricValue, ...],
)

capabilities instance-attribute 

capabilities: frozenset[str]

metrics instance-attribute 

metrics: tuple[MetricValue, ...]

sampled_at instance-attribute 

sampled_at: datetime

source_id instance-attribute 

source_id: str

empty classmethod 

empty(
    source_id: str,
    *,
    capabilities: frozenset[str] = frozenset(),
) -> TelemetrySnapshot

TelemetrySource

Bases: Protocol

A read-only source of normalized telemetry snapshots.

source_id instance-attribute 

source_id: str

probe async 

probe() -> SourceAvailability

sample async 

sample() -> TelemetrySnapshot

Bounded sampler for telemetry sources.

TelemetryRuntime 

TelemetryRuntime(
    sources: Iterable[TelemetrySource],
    *,
    timeout_seconds: float = 2.0,
    history_limit: int = 120,
)

Collect snapshots without coupling callers to provider classes or Textual.

sources instance-attribute 

sources = tuple(sources)

timeout_seconds instance-attribute 

timeout_seconds = timeout_seconds

history 

history(source_id: str) -> tuple[TelemetrySnapshot, ...]

probe_all async 

probe_all() -> Mapping[str, SourceAvailability]

sample_all async 

sample_all() -> tuple[TelemetrySnapshot, ...]