Cookbook

On this page

Trace to multiple backends at once
Toggle tracing on and off without losing exporter setup
Capture spans in memory for assertions in tests
Bridge an existing, already-computed trace into this module
Writing a custom exporter
Instrumenting a function vs. wrapping a block

Practical recipes built directly from the module's actual API surface.

Trace to multiple backends at once

Python · 4 lines

from zerda.observability import get_tracer, ConsoleExporter, JSONExporter

get_tracer().register_exporter(ConsoleExporter())
get_tracer().register_exporter(JSONExporter(path="traces.jsonl"))

Every span now reaches both exporters (dispatched in registration order by the single background drain thread — see Architecture).

Toggle tracing on and off without losing exporter setup

Python · 8 lines

from zerda.observability import enable_console, disable_tracing, get_tracer

enable_console()
# ... traced section ...
disable_tracing()
# ... untraced section: Tracer.span() takes the cheap no-op path ...
get_tracer().enable()
# ... tracing resumes to the same ConsoleExporter, no re-registration needed ...

Capture spans in memory for assertions in tests

This mirrors the pattern the project's own test suite uses (a RecordingExporter subclassing BaseExporter):

Python · 19 lines

from zerda.observability import BaseExporter, Span, get_tracer

class RecordingExporter(BaseExporter):
    def __init__(self):
        self.spans = []

    def export(self, span: Span) -> None:
        self.spans.append(span)

recorder = RecordingExporter()
get_tracer().clear_exporters()   # isolate from any prior test/state
get_tracer().register_exporter(recorder)

with get_tracer().span("Step", type="tool") as span:
    span.set_output("done")

get_tracer().flush()             # block until the drain thread has processed it
assert recorder.spans[0].name == "Step"
assert recorder.spans[0].success is True

Or, without writing your own class, use the built-in JSONExporter in its in-memory mode for the same purpose:

Python · 10 lines

from zerda.observability import get_tracer, JSONExporter

exporter = JSONExporter()  # no path -> in-memory
get_tracer().register_exporter(exporter)

with get_tracer().span("Step", type="tool"):
    pass

get_tracer().flush()
assert exporter.spans[0]["name"] == "Step"

Bridge an existing, already-computed trace into this module

If some other subsystem already measured a span's timing end-to-end (rather than wrapping live code with with tracer.span(...)), report it via record_external_span() instead of trying to fabricate a with block around code that already ran:

Python · 13 lines

from zerda.observability import get_tracer

get_tracer().record_external_span(
    name="ExecutionTracer step",
    span_type="chain",
    start_time=step.start_ts,
    end_time=step.end_ts,
    input=step.input,
    output=step.output,
    error=step.error_message,   # a string, not an exception object
    trace_id=step.trace_id,     # pass explicitly — this is NOT read from the
    parent_id=step.parent_id,   # currently-open span's ContextVar
)

Writing a custom exporter

Subclass BaseExporter and implement export(). Keep it defensive even though Tracer also isolates failures — the contract expects it:

Python · 27 lines

import logging
from zerda.observability import BaseExporter, Span, get_tracer

logger = logging.getLogger(__name__)

class OpenTelemetryExporter(BaseExporter):
    """Sketch — adapt to your actual OTel SDK setup."""

    def __init__(self, otel_tracer):
        self._otel_tracer = otel_tracer

    def export(self, span: Span) -> None:
        try:
            with self._otel_tracer.start_as_current_span(span.name) as otel_span:
                otel_span.set_attribute("fennec.span_id", span.id)
                otel_span.set_attribute("fennec.trace_id", span.trace_id)
                otel_span.set_attribute("fennec.type", span.type)
                if span.error:
                    otel_span.set_status_error(span.error)
        except Exception:
            logger.exception("failed to export span %s to OpenTelemetry", span.id)

    def shutdown(self) -> None:
        # Flush/close whatever your OTel SDK setup requires.
        pass

get_tracer().register_exporter(OpenTelemetryExporter(my_otel_tracer))

[!NOTE] This example is a sketch showing where the integration point is (export() receiving a Fennec Span), not a verified integration — the actual OpenTelemetry SDK calls depend on which OTel Python API version you use, which is outside the scope of this module's source.

Instrumenting a function vs. wrapping a block

Prefer the trace() decorator when a whole function is the unit of work and you don't need to touch the span mid-function; prefer get_tracer().span(...) directly when you need to call span.set_input(...) with something other than the raw call arguments, or need to read/mutate the span partway through:

Python · 16 lines

from zerda.observability import trace, get_tracer

# Whole function is the unit of work:
@trace("EmbedQuery", type="embedding")
def embed_query(text: str) -> list[float]:
    return embedding_model.encode(text)

# Need finer control (custom input shape, mid-function metadata):
def embed_and_search(text: str):
    with get_tracer().span("EmbedAndSearch", type="retriever") as span:
        span.set_input({"query": text, "top_k": 5})
        vec = embedding_model.encode(text)
        span.set_metadata(embedding_dim=len(vec))
        results = vector_store.search(vec, top_k=5)
        span.set_output(results)
        return results