harnest.telemetry

All Python modules · Import and compatibility guidance

Exporter factory contract

Requirement	Behavior
Unique `TelemetryExporter.name`	Duplicate names stop runtime startup
Synchronous, zero-argument factory	Runs once during runtime bootstrap
`traces`, `logs`, or both	Metrics exporters are not supported
`order=`	Resolves factories in ascending lifecycle order
Startup failure	Already-created exporters are shut down

Environment controls

Variable	Effect
`OTEL_SERVICE_NAME`	Sets the service identity
`HARNEST_LOG_LEVEL`	Sets the minimum log level; defaults to `INFO`
`HARNEST_LOG_CONSOLE=false`	Disables JSON logs on stderr
`HARNEST_OTEL_ENABLED=false`	Disables the environment-configured OTLP sink
`OTEL_SDK_DISABLED=true`	Disables the environment-configured OTLP sink

Exports

Name	Kind
`TelemetryExporter`	Class
`TelemetryExporterError`	Class
`TelemetryState`	Class
`configure_observability`	Function
`get_tracer`	Function
`instrument_fastapi`	Function
`resolve_telemetry_exporters`	Function

TelemetryExporter

One named telemetry destination with optional trace and log exporters.

TelemetryExporter(name: str, traces: Any | None = None, logs: Any | None = None) -> None

Fields (— means no declared default):

Field	Type	Default
`name`	`str`	`—`
`traces`	`Any \| None`	`None`
`logs`	`Any \| None`	`None`

Parameters and example

Parameter	Type	Default
`name`	`str`	Required
`traces`	`Any \| None`	`None`
`logs`	`Any \| None`	`None`

from harnest.telemetry import TelemetryExporter

exporter = TelemetryExporter(name=...)

TelemetryExporterError

A runtime telemetry-exporter factory returned an invalid destination.

Example

from harnest.telemetry import TelemetryExporterError

try:
    ...
except TelemetryExporterError as error:
    ...

TelemetryState

Own the providers and processors configured for one agent runtime.

TelemetryState(
    service_name: str,
    framework: str,
    enabled: bool,
    tracer_provider: Any | None = None,
    logger_provider: Any | None = None,
    owns_tracer_provider: bool = False,
    owns_logger_provider: bool = False,
    span_processors: tuple[Any, ...] = (),
    log_processors: tuple[Any, ...] = (),
    shutdown_called: bool = False,
) -> None

Fields (— means no declared default):

Field	Type	Default
`service_name`	`str`	`—`
`framework`	`str`	`—`
`enabled`	`bool`	`—`
`tracer_provider`	`Any \| None`	`None`
`logger_provider`	`Any \| None`	`None`
`owns_tracer_provider`	`bool`	`False`
`owns_logger_provider`	`bool`	`False`
`span_processors`	`tuple[Any, ...]`	`()`
`log_processors`	`tuple[Any, ...]`	`()`
`shutdown_called`	`bool`	`False`

Parameters and example

Parameter	Type	Default
`service_name`	`str`	Required
`framework`	`str`	Required
`enabled`	`bool`	Required
`tracer_provider`	`Any \| None`	`None`
`logger_provider`	`Any \| None`	`None`
`owns_tracer_provider`	`bool`	`False`
`owns_logger_provider`	`bool`	`False`
`span_processors`	`tuple[Any, ...]`	`()`
`log_processors`	`tuple[Any, ...]`	`()`
`shutdown_called`	`bool`	`False`

from harnest.telemetry import TelemetryState

state = TelemetryState(service_name=..., framework=..., enabled=...)

TelemetryState.force_flush

TelemetryState.force_flush(timeout_millis: int = 5000) -> None

Flush owned providers or only processors added to adopted providers.

Parameters and example

Parameter	Type	Default
`timeout_millis`	`int`	`5000`

from harnest.telemetry import TelemetryState

state: TelemetryState
state.force_flush()

TelemetryState.shutdown

TelemetryState.shutdown() -> None

Shut down providers or processors created and owned by Harnest.

Example

from harnest.telemetry import TelemetryState

state: TelemetryState
state.shutdown()

configure_observability

Configure console logging and optional OTLP traces/logs once per process.

configure_observability(
    service_name: str,
    *,
    framework: str,
    service_version: str | None = None,
    span_exporter: Any | None = None,
    log_exporter: Any | None = None,
    exporters: Sequence[TelemetryExporter] = (),
    exporter_factories: Sequence[Any] = (),
    use_global_providers: bool = False,
    set_global_providers: bool = True,
) -> TelemetryState

Parameters and example

Parameter	Type	Default
`service_name`	`str`	Required
`framework`	`str`	Required
`service_version`	`str \| None`	`None`
`span_exporter`	`Any \| None`	`None`
`log_exporter`	`Any \| None`	`None`
`exporters`	`Sequence[TelemetryExporter]`	`()`
`exporter_factories`	`Sequence[Any]`	`()`
`use_global_providers`	`bool`	`False`
`set_global_providers`	`bool`	`True`

from harnest.telemetry import configure_observability

state = configure_observability(service_name=..., framework=...)

get_tracer

Return a tracer from Harnest’s configured or global telemetry provider.

get_tracer(name: str, *, version: str | None = None) -> Any

Parameters and example

Parameter	Type	Default
`name`	`str`	Required
`version`	`str \| None`	`None`

from harnest.telemetry import get_tracer

result = get_tracer(name=...)

instrument_fastapi

Attach server request spans to one FastAPI application.

instrument_fastapi(app: Any, state: TelemetryState) -> None

Parameters and example

Parameter	Type	Default
`app`	`Any`	Required
`state`	`TelemetryState`	Required

from harnest.telemetry import instrument_fastapi

instrument_fastapi(app=..., state=...)

resolve_telemetry_exporters

Call discovered exporter factories only during runtime bootstrap.

resolve_telemetry_exporters(factories: Sequence[Any]) -> tuple[TelemetryExporter, ...]

Parameters and example

Parameter	Type	Default
`factories`	`Sequence[Any]`	Required

from harnest.telemetry import resolve_telemetry_exporters

result = resolve_telemetry_exporters(factories=...)

Overview

Get started

Harnest Studio

Guides

API reference

harnest.telemetry

Exporter factory contract

Environment controls

Exports

TelemetryExporter

TelemetryExporterError

TelemetryState

TelemetryState.force_flush

TelemetryState.shutdown

configure_observability

get_tracer

instrument_fastapi

resolve_telemetry_exporters

​Exporter factory contract

​Environment controls

​Exports

​TelemetryExporter

​TelemetryExporterError

​TelemetryState

​TelemetryState.force_flush

​TelemetryState.shutdown

​configure_observability

​get_tracer

​instrument_fastapi

​resolve_telemetry_exporters

Exporter factory contract

Environment controls

Exports

TelemetryExporter

TelemetryExporterError

TelemetryState

TelemetryState.force_flush

TelemetryState.shutdown

configure_observability

get_tracer

instrument_fastapi

resolve_telemetry_exporters