Coverage for dataexcept/sentry.py: 80%

1"""Enrich Sentry error events with structured DataExcept metadata.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Mapping 

6from typing import Any 

7 

8from .observability import OperationContext 

9from .serialization import exception_to_dict 

10 

11__all__ = ["enrich_sentry_event"] 

12 

13 

14def _exception_from_hint(hint: Mapping[str, Any]) -> BaseException | None: 

15 try: 

16 exc_info = hint.get("exc_info") 

17 except Exception: # pragma: no cover - hostile mapping implementations 

18 return None 

19 

20 if not isinstance(exc_info, tuple) or len(exc_info) < 2: 

21 return None 

22 

23 exc = exc_info[1] 

24 return exc if isinstance(exc, BaseException) else None 

25 

26 

27def _tags_from_envelope(envelope: Mapping[str, Any]) -> dict[str, str]: 

28 tags: dict[str, str] = {} 

29 

30 exception_type = envelope.get("type") 

31 if isinstance(exception_type, str): 31 ↛ 34line 31 didn't jump to line 34 because the condition on line 31 was always true

32 tags["dataexcept.type"] = exception_type 

33 

34 failure = envelope.get("failure") 

35 if not isinstance(failure, Mapping): 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true

36 return tags 

37 

38 kind = failure.get("kind") 

39 if isinstance(kind, str): 39 ↛ 42line 39 didn't jump to line 42 because the condition on line 39 was always true

40 tags["dataexcept.failure_kind"] = kind 

41 

42 retryable = failure.get("retryable") 

43 if isinstance(retryable, bool): 43 ↛ 45line 43 didn't jump to line 45 because the condition on line 43 was always true

44 tags["dataexcept.retryable"] = str(retryable).lower() 

45 elif retryable is None: 

46 tags["dataexcept.retryable"] = "unknown" 

47 

48 return tags 

49 

50 

51def _operation_tags(operation_context: OperationContext | None) -> dict[str, str]: 

52 if operation_context is None: 

53 return {} 

54 if not isinstance(operation_context, OperationContext): 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true

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

56 return { 

57 f"dataexcept.operation.{key}": value 

58 for key, value in operation_context.index_fields().items() 

59 } 

60 

61 

62def enrich_sentry_event( 

63 event: dict[str, Any], 

64 hint: Mapping[str, Any], 

65 *, 

66 operation_context: OperationContext | None = None, 

67 include_attributes: bool = True, 

68 max_depth: int = 8, 

69) -> dict[str, Any]: 

70 """Return a Sentry event enriched with DataExcept failure metadata. 

71 

72 Full operation context, including correlation identifiers, is stored under 

73 ``contexts.dataexcept_operation``. Only the low-cardinality operation fields 

74 become tags so request/job/trace identifiers do not become indexed tags. 

75 """ 

76 if not isinstance(event, dict): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 raise TypeError("event must be a dictionary") 

78 if not isinstance(hint, Mapping): 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 raise TypeError("hint must be a mapping") 

80 if operation_context is not None and not isinstance( 80 ↛ 83line 80 didn't jump to line 83 because the condition on line 80 was never true

81 operation_context, OperationContext 

82 ): 

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

84 

85 exc = _exception_from_hint(hint) 

86 if exc is None: 

87 return event 

88 

89 envelope = exception_to_dict( 

90 exc, 

91 include_attributes=include_attributes, 

92 max_depth=max_depth, 

93 ) 

94 enriched = dict(event) 

95 

96 existing_contexts = event.get("contexts") 

97 contexts = dict(existing_contexts) if isinstance(existing_contexts, Mapping) else {} 

98 contexts["dataexcept"] = envelope 

99 if operation_context is not None: 

100 serialized = operation_context.to_dict() 

101 if serialized: 101 ↛ 103line 101 didn't jump to line 103 because the condition on line 101 was always true

102 contexts["dataexcept_operation"] = serialized 

103 enriched["contexts"] = contexts 

104 

105 existing_tags = event.get("tags") 

106 tags = dict(existing_tags) if isinstance(existing_tags, Mapping) else {} 

107 tags.update(_tags_from_envelope(envelope)) 

108 tags.update(_operation_tags(operation_context)) 

109 enriched["tags"] = tags 

110 

111 return enriched