Coverage for src / lexigram / contracts / observability / tracing.py: 0%

1"""OpenTelemetry protocols for distributed tracing. 

2 

3This module defines protocols for OpenTelemetry-compatible tracing interfaces. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

9 

10if TYPE_CHECKING: 

11 from typing import Self 

12 

13 

14@runtime_checkable 

15class TracerProtocol(Protocol): 

16 """Protocol for OpenTelemetry-compatible tracers.""" 

17 

18 def start_span( 

19 self, 

20 name: str, 

21 attributes: dict[str, Any] | None = None, 

22 context: Any | None = None, 

23 ) -> SpanProtocol: 

24 """Start a new span. 

25 

26 Args: 

27 name: Span name 

28 attributes: Optional span attributes 

29 context: Optional parent context (from extract_context) 

30 

31 Returns: 

32 New span instance 

33 """ 

34 ... 

35 

36 def get_current_span(self) -> SpanProtocol | None: 

37 """Get the currently active span. 

38 

39 Returns: 

40 Current span or None 

41 """ 

42 ... 

43 

44 def inject_context( 

45 self, carrier: dict[str, str], context: Any | None = None 

46 ) -> None: 

47 """Inject trace context into a carrier dict (W3C traceparent). 

48 

49 Implementations should write a ``traceparent`` key (and optionally 

50 ``tracestate``) into *carrier* so that downstream consumers can 

51 continue the distributed trace. 

52 

53 Args: 

54 carrier: Mutable dict that will receive the trace headers. 

55 context: Optional explicit context to inject. If None, uses the 

56 current active span's context. 

57 """ 

58 ... 

59 

60 def extract_context(self, carrier: dict[str, str]) -> Any | None: 

61 """Extract trace context from a carrier dict (W3C traceparent). 

62 

63 Args: 

64 carrier: Dict containing trace headers (e.g. message headers). 

65 

66 Returns: 

67 An opaque context object suitable for passing to ``start_span``, 

68 or *None* if no valid context was found. 

69 """ 

70 ... 

71 

72 

73@runtime_checkable 

74class SpanProtocol(Protocol): 

75 """Protocol for OpenTelemetry-compatible spans.""" 

76 

77 def set_attribute(self, key: str, value: Any) -> None: 

78 """Set a span attribute. 

79 

80 Args: 

81 key: Attribute key 

82 value: Attribute value 

83 """ 

84 ... 

85 

86 def add_event(self, name: str, attributes: dict[str, Any] | None = None) -> None: 

87 """Add an event to the span. 

88 

89 Args: 

90 name: Event name 

91 attributes: Optional event attributes 

92 """ 

93 ... 

94 

95 def record_exception(self, exception: Exception) -> None: 

96 """Record an exception on the span. 

97 

98 Args: 

99 exception: Exception to record 

100 """ 

101 ... 

102 

103 def set_status(self, status: str) -> None: 

104 """Set the span status. 

105 

106 Args: 

107 status: Status string 

108 """ 

109 ... 

110 

111 def __enter__(self) -> Self: 

112 """Enter the span context (for 'with' statement).""" 

113 ... 

114 

115 def __exit__( 

116 self, 

117 exc_type: type[BaseException] | None, 

118 exc_val: BaseException | None, 

119 exc_tb: object, 

120 ) -> None: 

121 """Exit the span context.""" 

122 ... 

123 

124 def end(self) -> None: 

125 """Mark span as finished. 

126 

127 Called automatically by __exit__, but can be called explicitly 

128 for early ending of a span. 

129 """ 

130 ... 

131 

132 

133__all__ = ["SpanProtocol", "TracerProtocol"]