Coverage for dataexcept/opentelemetry.py: 92%
68 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-27 14:44 +0000
1"""OpenTelemetry-compatible exception attributes without an OTel dependency.
3OpenTelemetry defines stable ``exception.type``, ``exception.message`` and
4``exception.stacktrace`` attributes for exceptions. DataExcept can supply those
5values from its redacted envelope, plus its recovery metadata and product-neutral
6operation context, without importing ``opentelemetry`` itself.
7"""
9from __future__ import annotations
11import json
12import traceback
13from collections.abc import Mapping
14from typing import Protocol
16from .observability import OperationContext
17from .redaction import redact_urls_in_text
18from .schema import ENVELOPE_SCHEMA_ID
19from .serialization import exception_to_dict
21__all__ = [
22 "ExceptionRecorder",
23 "OtelAttributeValue",
24 "exception_to_otel_attributes",
25 "record_otel_exception",
26]
28OtelAttributeValue = str | bool | int | float
31class ExceptionRecorder(Protocol):
32 """Minimal structural interface required by :func:`record_otel_exception`."""
34 def record_exception(
35 self,
36 exception: BaseException,
37 attributes: Mapping[str, OtelAttributeValue] | None = None,
38 ) -> None:
39 """Record *exception* with optional event attributes."""
42def _qualified_type_name(exc: BaseException) -> str:
43 cls = type(exc)
44 return f"{cls.__module__}.{cls.__qualname__}"
47def _rendered_stacktrace(exc: BaseException) -> str | None:
48 if exc.__traceback__ is None:
49 return None
50 try:
51 rendered = "".join(
52 traceback.format_exception(type(exc), exc, exc.__traceback__)
53 )
54 except Exception: # pragma: no cover - hostile traceback objects
55 return None
56 return redact_urls_in_text(rendered, keep_path=False) or None
59def _failure_attributes(
60 envelope: Mapping[str, object],
61) -> dict[str, OtelAttributeValue]:
62 failure = envelope.get("failure")
63 if not isinstance(failure, Mapping): 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true
64 return {}
66 attributes: dict[str, OtelAttributeValue] = {}
67 kind = failure.get("kind")
68 if isinstance(kind, str): 68 ↛ 71line 68 didn't jump to line 71 because the condition on line 68 was always true
69 attributes["dataexcept.failure.kind"] = kind
71 retryable = failure.get("retryable")
72 if isinstance(retryable, bool): 72 ↛ 75line 72 didn't jump to line 75 because the condition on line 72 was always true
73 attributes["dataexcept.failure.retryable"] = retryable
75 retry_after = failure.get("retry_after_seconds")
76 if isinstance(retry_after, (int, float)) and not isinstance(retry_after, bool): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true
77 attributes["dataexcept.failure.retry_after_seconds"] = float(retry_after)
79 return attributes
82def _operation_attributes(
83 operation_context: OperationContext | None,
84) -> dict[str, OtelAttributeValue]:
85 """Project operation context without duplicating native trace identifiers."""
86 if operation_context is None:
87 return {}
88 if not isinstance(operation_context, OperationContext):
89 raise TypeError("operation_context must be an OperationContext or None")
91 attributes: dict[str, OtelAttributeValue] = {}
92 values = operation_context.to_dict()
93 for key in (
94 "system",
95 "component",
96 "operation",
97 "request_id",
98 "job_id",
99 "correlation_id",
100 ):
101 value = values.get(key)
102 if value is not None: 102 ↛ 93line 102 didn't jump to line 93 because the condition on line 102 was always true
103 attributes[f"dataexcept.operation.{key}"] = value
104 return attributes
107def exception_to_otel_attributes(
108 exc: BaseException,
109 *,
110 operation_context: OperationContext | None = None,
111 include_attributes: bool = True,
112 max_depth: int = 8,
113 include_stacktrace: bool = True,
114 include_envelope: bool = False,
115) -> dict[str, OtelAttributeValue]:
116 """Return OpenTelemetry-compatible attributes describing *exc*.
118 ``trace_id`` and ``span_id`` from :class:`OperationContext` are deliberately
119 not duplicated as custom attributes. OpenTelemetry already carries them in
120 native span context; DataExcept only projects operation and correlation data.
121 """
122 envelope = exception_to_dict(
123 exc,
124 include_attributes=include_attributes,
125 max_depth=max_depth,
126 )
128 attributes: dict[str, OtelAttributeValue] = {
129 "exception.type": _qualified_type_name(exc),
130 "exception.message": str(envelope.get("message", "")),
131 }
133 if include_stacktrace:
134 stacktrace = _rendered_stacktrace(exc)
135 if stacktrace is not None:
136 attributes["exception.stacktrace"] = stacktrace
138 attributes.update(_failure_attributes(envelope))
139 attributes.update(_operation_attributes(operation_context))
141 if include_envelope:
142 attributes["dataexcept.envelope.schema"] = ENVELOPE_SCHEMA_ID
143 attributes["dataexcept.envelope"] = json.dumps(
144 envelope,
145 ensure_ascii=False,
146 allow_nan=False,
147 separators=(",", ":"),
148 sort_keys=True,
149 )
151 return attributes
154def record_otel_exception(
155 span: ExceptionRecorder,
156 exc: BaseException,
157 *,
158 operation_context: OperationContext | None = None,
159 include_attributes: bool = True,
160 max_depth: int = 8,
161 include_stacktrace: bool = True,
162 include_envelope: bool = False,
163) -> None:
164 """Record *exc* without allowing telemetry failure to escape.
166 Attribute conversion remains strict through
167 :func:`exception_to_otel_attributes`. This emission helper is different:
168 it is intended for use while handling an existing failure, so conversion or
169 recorder errors are swallowed rather than replacing that failure.
170 """
171 try:
172 attributes = exception_to_otel_attributes(
173 exc,
174 operation_context=operation_context,
175 include_attributes=include_attributes,
176 max_depth=max_depth,
177 include_stacktrace=include_stacktrace,
178 include_envelope=include_envelope,
179 )
180 span.record_exception(exc, attributes=attributes)
181 except Exception:
182 return