Coverage for src/lexigram/web/middleware/request_id.py: 35%

1"""Request ID middleware for request tracing. 

2 

3Adds unique request ID to each request and propagates through logs. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any 

9import uuid 

10 

11from starlette.requests import Request 

12from starlette.types import ASGIApp, Receive, Scope, Send 

13 

14from lexigram.contracts.core.identity import IdGeneratorProtocol 

15from lexigram.logging import get_logger 

16from lexigram.primitives.context import REQUEST_ID, Context 

17 

18logger = get_logger(__name__) 

19 

20 

21class RequestIDMiddleware: 

22 """Pure ASGI middleware to add request ID to each request. 

23 

24 10x faster than BaseHTTPMiddleware - no task creation overhead. 

25 

26 Features: 

27 - Generates unique request ID for each request 

28 - Accepts existing request ID from X-Request-ID header 

29 - Adds request ID to response headers 

30 - Propagates request ID to context for logging 

31 

32 Example: 

33 >>> from lexigram.web.middleware import RequestIDMiddleware 

34 >>> 

35 >>> # Request 

36 >>> GET /api/users/123 

37 >>> # (no X-Request-ID header) 

38 >>> 

39 >>> # Response 

40 >>> X-Request-ID: req_a1b2c3d4 

41 >>> 

42 >>> # Logs 

43 >>> 2026-01-12 10:23:15 INFO [req_a1b2c3d4] Fetching user 123 

44 """ 

45 

46 def __init__( 

47 self, 

48 app: ASGIApp, 

49 ctx: Context | None = None, 

50 header_name: str = "X-Request-ID", 

51 ids: IdGeneratorProtocol | None = None, 

52 ): 

53 """Initialize request ID middleware. 

54 

55 Args: 

56 app: ASGI application 

57 ctx: Application context for propagating the request ID. 

58 header_name: Header name for request ID (default: X-Request-ID). 

59 ids: ID generator protocol for request ID generation. 

60 """ 

61 self.app = app 

62 self._ctx = ctx 

63 self.header_name = header_name 

64 self._ids = ids 

65 

66 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

67 """Pure ASGI middleware implementation.""" 

68 

69 if scope["type"] != "http": 

70 # Pass through non-HTTP requests 

71 await self.app(scope, receive, send) 

72 return 

73 

74 # Get or generate request ID from headers 

75 headers = dict(scope.get("headers", [])) 

76 header_bytes = self.header_name.lower().encode() 

77 request_id_bytes = headers.get(header_bytes) 

78 

79 if request_id_bytes: 

80 request_id = request_id_bytes.decode() 

81 else: 

82 # Generate new UUID-based request ID 

83 request_id = ( 

84 f"req_{self._ids.generate() if self._ids else uuid.uuid4().hex[:12]}" 

85 ) 

86 

87 # Set in context (for logging and background tasks) 

88 if self._ctx is not None: 

89 self._ctx.set(REQUEST_ID, request_id) 

90 

91 # Store request ID in scope for access in handlers 

92 scope["request_id"] = request_id 

93 

94 logger.debug( 

95 "Request ID assigned", 

96 request_id=request_id, 

97 method=scope.get("method", ""), 

98 path=scope.get("path", ""), 

99 ) 

100 

101 # Wrap send to add request ID header to response 

102 async def send_with_request_id(message: Any) -> None: 

103 if message["type"] == "http.response.start": 

104 headers = list(message.get("headers", [])) 

105 # Add request ID header 

106 headers.append([header_bytes, request_id.encode()]) 

107 message["headers"] = headers 

108 await send(message) 

109 

110 # Process request 

111 await self.app(scope, receive, send_with_request_id) 

112 

113 

114def get_request_id_from_request(request: Request) -> str | None: 

115 """Get request ID from request state. 

116 

117 Args: 

118 request: Lexigram request. 

119 

120 Returns: 

121 Request ID or None. 

122 

123 Example: 

124 >>> @app.get("/api/status") 

125 >>> async def get_status(request: Request): 

126 ... request_id = get_request_id_from_request(request) 

127 ... return {"request_id": request_id} 

128 """ 

129 return getattr(request.state, "request_id", None) 

130 

131 

132class RequestIDLogFilter: 

133 """Log filter to add request ID to log records. 

134 

135 This filter extracts the current request ID from the injected Context 

136 and adds it to each log record as `request_id`. 

137 """ 

138 

139 def __init__(self, ctx: Context | None = None) -> None: 

140 """Initialize filter. 

141 

142 Args: 

143 ctx: Application context instance. If provided, the request ID 

144 is read via ``ctx.get(REQUEST_ID)``. 

145 """ 

146 self._ctx = ctx 

147 

148 def filter(self, record: object) -> bool: 

149 """Add request ID to log record.""" 

150 request_id = self._ctx.get(REQUEST_ID) if self._ctx is not None else None 

151 record.request_id = request_id or "no_request" # type: ignore[attr-defined] 

152 return True 

153 

154 

155def configure_logging_with_request_id() -> None: 

156 """Configure standard logging to include request ID. 

157 

158 This adds a filter to the root logger to ensure all logs include the 

159 current request ID if available. 

160 """ 

161 

162 root_logger = get_logger() 

163 if not any(isinstance(f, RequestIDLogFilter) for f in root_logger.filters): 

164 root_logger.addFilter(RequestIDLogFilter())