Telemetry

Headless source, metric, snapshot, and sampling contracts. See Telemetry for the boundary between reusable infrastructure measurements and application-owned domain metrics.

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, ...]