Coverage for src/lexigram/web/interceptors/builtin/logging.py: 26%

1"""Logging interceptor for request/response logging. 

2 

3Provides structured logging for HTTP requests and responses. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any, cast 

9 

10from lexigram.logging import get_logger 

11from lexigram.primitives import clock as ambient_clock 

12from lexigram.web.protocols import ( 

13 CallHandlerProtocol, 

14 ExecutionContextProtocol, 

15 WebInterceptorProtocol, 

16) 

17 

18logger = get_logger(__name__) 

19 

20 

21class LoggingInterceptor(WebInterceptorProtocol): 

22 """Interceptor that logs request and response information. 

23 

24 **Development/debug only** — not intended for production use. 

25 For production request logging, use ``AccessLogMiddleware`` instead. 

26 

27 Logs the request method, path, status code, and elapsed time 

28 for each HTTP request. 

29 

30 Example: 

31 ```python 

32 # Add to global interceptors 

33 router.add_interceptor(LoggingInterceptor()) 

34 

35 # Or use decorator 

36 @use_interceptors(LoggingInterceptor()) 

37 class UserController(Controller): 

38 ... 

39 ``` 

40 """ 

41 

42 def __init__( 

43 self, 

44 log_request: bool = True, 

45 log_response: bool = True, 

46 log_body: bool = False, 

47 logger_name: str | None = None, 

48 ): 

49 """Initialize the logging interceptor. 

50 

51 Args: 

52 log_request: Whether to log incoming requests. 

53 log_response: Whether to log responses. 

54 log_body: Whether to log request/response bodies (careful with sensitive data). 

55 logger_name: Custom logger name to use. 

56 """ 

57 self._log_request = log_request 

58 self._log_response = log_response 

59 self._log_body = log_body 

60 self._logger = get_logger(logger_name or __name__) 

61 

62 async def intercept( 

63 self, 

64 context: ExecutionContextProtocol, 

65 next_handler: CallHandlerProtocol, 

66 ) -> Any: 

67 """Intercept the request and log details. 

68 

69 Args: 

70 context: The execution context with request info. 

71 next_handler: The next handler in the chain. 

72 

73 Returns: 

74 The response from the handler. 

75 """ 

76 request = context.request 

77 method = getattr(request, "method", "UNKNOWN") 

78 url = getattr(request, "url", None) 

79 path = url.path if url and hasattr(url, "path") else "/" 

80 

81 # Log incoming request 

82 if self._log_request: 

83 extra = { 

84 "method": method, 

85 "path": path, 

86 "client": self._get_client_ip(request), 

87 } 

88 

89 if self._log_body and hasattr(request, "body"): 

90 # Don't await here to not block 

91 extra["body_size"] = request.headers.get("content-length", "unknown") 

92 

93 cast("Any", self._logger).info( 

94 "Request: %s %s", 

95 method, 

96 path, 

97 extra=extra, 

98 ) 

99 

100 # Track timing 

101 start_time = ambient_clock.monotonic() 

102 

103 try: 

104 # Call the next handler 

105 result = await next_handler.handle() 

106 

107 # Calculate elapsed time 

108 elapsed = (ambient_clock.monotonic() - start_time) * 1000 # ms 

109 

110 # Log response 

111 if self._log_response: 

112 status_code = getattr(result, "status_code", 200) if result else 500 

113 cast("Any", self._logger).info( 

114 "Response: %s %s - %s (%.1fms)", 

115 method, 

116 path, 

117 status_code, 

118 elapsed, 

119 extra={ 

120 "method": method, 

121 "path": path, 

122 "status_code": status_code, 

123 "elapsed_ms": elapsed, 

124 }, 

125 ) 

126 

127 return result 

128 

129 except Exception as e: 

130 # Log error 

131 elapsed = (ambient_clock.monotonic() - start_time) * 1000 

132 cast("Any", self._logger).error( 

133 "Error: %s %s - %s: %s (%.1fms)", 

134 method, 

135 path, 

136 type(e).__name__, 

137 e, 

138 elapsed, 

139 extra={ 

140 "method": method, 

141 "path": path, 

142 "error": str(e), 

143 "error_type": type(e).__name__, 

144 "elapsed_ms": elapsed, 

145 }, 

146 exc_info=True, 

147 ) 

148 raise 

149 

150 def _get_client_ip(self, request: Any) -> str: 

151 """Extract client IP from request. 

152 

153 Args: 

154 request: The HTTP request. 

155 

156 Returns: 

157 The client IP address. 

158 """ 

159 # Check X-Forwarded-For header 

160 forwarded = request.headers.get("x-forwarded-for") 

161 if forwarded: 

162 return cast("str", forwarded.split(",")[0].strip()) 

163 

164 # Fall back to direct client 

165 client = getattr(request, "client", None) 

166 if client: 

167 return cast("str", client.host) 

168 return "unknown" 

169 

170 

171__all__ = ["LoggingInterceptor"]