Coverage for dataexcept/observability.py: 98%

1"""Product-neutral observability context for failures crossing boundaries. 

2 

3Frameworks call the same concepts by different names: request, task, job, 

4workflow step, tool call, invocation. DataExcept keeps the shared part small 

5and explicit so adapters can project it into logs, traces, error trackers or 

6protocol-specific metadata without making those frameworks runtime 

7requirements. 

8""" 

9 

10from __future__ import annotations 

11 

12from dataclasses import dataclass, fields 

13from typing import Any 

14 

15from .redaction import redact_urls_in_text 

16from .serialization import exception_to_dict 

17 

18__all__ = ["OperationContext", "exception_to_observability_event"] 

19 

20 

21@dataclass(frozen=True, slots=True) 

22class OperationContext: 

23 """Identifiers that describe where an operation is executing. 

24 

25 ``system``, ``component`` and ``operation`` are intended to stay 

26 low-cardinality and are suitable for filtering. Request, job, correlation, 

27 trace and span identifiers are correlation data and should not normally be 

28 turned into metric dimensions or indexed tags. 

29 

30 Every value is optional because inventing provenance is worse than omitting 

31 it. URL-shaped values are redacted before they can leave this object. 

32 """ 

33 

34 system: str | None = None 

35 component: str | None = None 

36 operation: str | None = None 

37 request_id: str | None = None 

38 job_id: str | None = None 

39 correlation_id: str | None = None 

40 trace_id: str | None = None 

41 span_id: str | None = None 

42 

43 def __post_init__(self) -> None: 

44 for field in fields(self): 

45 value = getattr(self, field.name) 

46 if value is None: 

47 continue 

48 if not isinstance(value, str): 

49 raise TypeError(f"{field.name} must be a string or None") 

50 if not value.strip(): 

51 raise ValueError(f"{field.name} must not be empty") 

52 object.__setattr__( 

53 self, 

54 field.name, 

55 redact_urls_in_text(value, keep_path=False), 

56 ) 

57 

58 def to_dict(self) -> dict[str, str]: 

59 """Return only the identifiers that are actually present.""" 

60 return { 

61 field.name: value 

62 for field in fields(self) 

63 if (value := getattr(self, field.name)) is not None 

64 } 

65 

66 def index_fields(self) -> dict[str, str]: 

67 """Return only low-cardinality fields suitable for filtering.""" 

68 return { 

69 key: value 

70 for key in ("system", "component", "operation") 

71 if (value := getattr(self, key)) is not None 

72 } 

73 

74 

75def exception_to_observability_event( 

76 exc: BaseException, 

77 *, 

78 operation_context: OperationContext | None = None, 

79 include_attributes: bool = True, 

80 max_depth: int = 8, 

81) -> dict[str, Any]: 

82 """Return a strict-JSON-safe failure event shared by observability adapters. 

83 

84 The exception payload is the canonical DataExcept envelope. Operation 

85 metadata is separate so adapters can index stable fields while retaining 

86 high-cardinality correlation identifiers without flattening them into tags. 

87 """ 

88 if operation_context is not None and not isinstance( 

89 operation_context, OperationContext 

90 ): 

91 raise TypeError("operation_context must be an OperationContext or None") 

92 

93 event: dict[str, Any] = { 

94 "event": "exception", 

95 "exception": exception_to_dict( 

96 exc, 

97 include_attributes=include_attributes, 

98 max_depth=max_depth, 

99 ), 

100 } 

101 

102 if operation_context is not None: 

103 operation = operation_context.to_dict() 

104 if operation: 104 ↛ 107line 104 didn't jump to line 107 because the condition on line 104 was always true

105 event["operation"] = operation 

106 

107 return event