Skip to content

Trace contract validation with OpenTelemetry

pip install pycharter[otel]

Only opentelemetry-api is installed — bring your own SDK and exporter (the standard thin-instrumentation pattern). Without a configured SDK the API no-ops, so the instruments are always safe to attach.

Trace Validator calls

from pycharter import Validator
from pycharter.contrib.opentelemetry import instrument_validator

validator = instrument_validator(Validator.from_file("contracts/orders.yaml"))

result = validator.validate(record)
# span "pycharter.validate" with pycharter.contract.title /
# pycharter.contract.version / pycharter.valid attributes

results = validator.validate_batch(batch)
# span "pycharter.validate_batch" with pycharter.record_count /
# pycharter.valid_count

The wrapper delegates everything else to the original instance, so it drops in anywhere a Validator is used — including the FastAPI gate:

from pycharter.contrib.fastapi import contract_dependency

validate_order = contract_dependency(
    instrument_validator(Validator.from_file("contracts/orders.yaml"))
)

Trace streaming outcomes

OTelStreamObserver implements the streaming engine's observer protocol: one span per processed record, named pycharter.stream.record, with stream / outcome / contract attributes. When the source message carries a W3C traceparent header the span is parented into the producer's trace, and the span's duration reflects the time spent inside the validator:

from pycharter.contrib.opentelemetry import OTelStreamObserver
from pycharter.streaming import ContractStreamProcessor

processor = ContractStreamProcessor(
    ...,
    observers=[OTelStreamObserver()],
)

Pair with the engine's Prometheus metrics (Emit a quality score to Prometheus) for metrics + traces from the same pipeline.