1from __future__ import annotations
 2
 3from dataclasses import dataclass
 4
 5from opentelemetry import trace
 6from opentelemetry.trace import format_trace_id
 7
 8
 9@dataclass(frozen=True)
10class CurrentTrace:
11    """The active request's OpenTelemetry trace, as connect needs it.
12
13    `trace_id` is the 32-char hex id, or "" when there is no valid trace —
14    connect only installs an SDK tracer provider when export is configured,
15    so without a token requests get no-op spans with an all-zero id.
16
17    `sampled` is the final sampling decision (so a sub-1.0
18    CONNECT_TRACE_SAMPLE_RATE is accounted for here for free), or None when
19    there is no valid trace.
20    """
21
22    trace_id: str
23    sampled: bool | None
24
25
26def current_trace() -> CurrentTrace:
27    """Read the active span's trace id and sampling decision."""
28    span_context = trace.get_current_span().get_span_context()
29    if not span_context.is_valid:
30        return CurrentTrace(trace_id="", sampled=None)
31    return CurrentTrace(
32        trace_id=format_trace_id(span_context.trace_id),
33        sampled=span_context.trace_flags.sampled,
34    )