| """ | |
| Telemetry helpers (structured events + optional spans). | |
| This keeps OpenTelemetry optional so unit tests and lightweight installs don't | |
| need to pull in tracing dependencies. | |
| """ | |
| from __future__ import annotations | |
| from contextlib import contextmanager | |
| from typing import Any, Dict, Iterator, Optional | |
| def _try_get_tracer(): | |
| try: | |
| from opentelemetry import trace # type: ignore | |
| return trace.get_tracer("ylff") | |
| except Exception: # pragma: no cover | |
| return None | |
| def span(name: str, *, attributes: Optional[Dict[str, Any]] = None) -> Iterator[None]: | |
| """ | |
| Context manager that creates an OpenTelemetry span if available, | |
| otherwise a no-op. | |
| """ | |
| tracer = _try_get_tracer() | |
| if tracer is None: | |
| yield | |
| return | |
| with tracer.start_as_current_span(name) as s: # pragma: no cover | |
| if attributes: | |
| for k, v in attributes.items(): | |
| try: | |
| s.set_attribute(k, v) | |
| except Exception: | |
| # Avoid blowing up the business logic on bad attribute types. | |
| pass | |
| yield | |