Skip to content

pydantable.observe

Plan observation hooks for debugging and telemetry.

pydantable.observe

Opt-in observability hooks for pydantable execution and I/O boundaries.

span

Context manager for timing an operation and emitting one event.

Source code in python/pydantable/observe.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class span:
    """Context manager for timing an operation and emitting one event."""

    def __init__(self, op: str, **fields: Any) -> None:
        self._op = op
        self._fields = fields
        self._t0: float | None = None

    def __enter__(self) -> span:
        self._t0 = _now()
        return self

    def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
        t0 = self._t0 or _now()
        dt_ms = (_now() - t0) * 1000.0
        event = dict(self._fields)
        event.update(
            {
                "op": self._op,
                "duration_ms": dt_ms,
                "ok": exc_type is None,
            }
        )
        if exc_type is not None:
            event["error_type"] = getattr(exc_type, "__name__", str(exc_type))
        emit(event)

set_observer 

set_observer(fn)

Set a global observer callback (or disable with None).

Source code in python/pydantable/observe.py 
18
19
20
21
def set_observer(fn: Observer | None) -> None:
    """Set a global observer callback (or disable with `None`)."""
    global _OBSERVER
    _OBSERVER = fn

get_observer 

get_observer()

Return the callback set by :func:set_observer, or None if disabled.

Source code in python/pydantable/observe.py 
24
25
26
def get_observer() -> Observer | None:
    """Return the callback set by :func:`set_observer`, or ``None`` if disabled."""
    return _OBSERVER

emit 

emit(event)

Emit an event to the observer if configured, otherwise no-op.

If PYDANTABLE_TRACE is truthy and no observer is configured, events are still emitted to stderr via print(...) for a minimal default.

Source code in python/pydantable/observe.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def emit(event: dict[str, Any]) -> None:
    """
    Emit an event to the observer if configured, otherwise no-op.

    If `PYDANTABLE_TRACE` is truthy and no observer is configured, events are
    still emitted to stderr via `print(...)` for a minimal default.
    """
    obs = _OBSERVER
    if obs is not None:
        obs(event)
        return
    if trace_enabled():
        # Keep this intentionally simple and stdlib-only.
        print(f"pydantable.trace {event}")