Coverage for src/lexigram/web/middleware/access_log.py: 28%

1"""Access log middleware for HTTP request/response logging. 

2 

3Logs every HTTP request with structured context: method, path, status code, 

4duration, and request ID. Uses pure ASGI implementation for minimal overhead. 

5 

6Example output (dev mode): 

7 2026-01-15 10:23:15 [info] request_completed method=GET path=/api/users 

8 status=200 duration_ms=12.3 request_id=req_a1b2c3d4 

9 

10Example output (JSON mode): 

11 {"event": "request_completed", "method": "GET", "path": "/api/users", 

12 "status": 200, "duration_ms": 12.3, "request_id": "req_a1b2c3d4", ...} 

13""" 

14 

15from __future__ import annotations 

16 

17from typing import TYPE_CHECKING, Any, cast 

18 

19from lexigram.logging import get_logger 

20from lexigram.primitives import clock as ambient_clock 

21from lexigram.primitives.context import REQUEST_ID, Context 

22from lexigram.web import constants as const 

23 

24if TYPE_CHECKING: 

25 from collections.abc import MutableMapping 

26 

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

28 

29logger = get_logger(__name__) 

30 

31 

32class AccessLogMiddleware: 

33 """Pure ASGI middleware for structured HTTP access logging. 

34 

35 Features: 

36 - Logs method, path, status code, and duration for every request 

37 - Automatically includes request_id from context 

38 - Zero overhead for non-HTTP scopes (websocket, lifespan) 

39 - Structured output compatible with JSON and console renderers 

40 

41 Example:: 

42 

43 from lexigram.web.middleware.access_log import AccessLogMiddleware 

44 

45 app.add_middleware(AccessLogMiddleware) 

46 """ 

47 

48 def __init__( 

49 self, 

50 app: ASGIApp, 

51 ctx: Context | None = None, 

52 *, 

53 exclude_paths: list[str] | None = None, 

54 ) -> None: 

55 """Initialize access log middleware. 

56 

57 Args: 

58 app: ASGI application. 

59 ctx: Application context for reading request metadata. 

60 exclude_paths: Paths to exclude from logging (e.g., ["/health"]). 

61 """ 

62 self.app = app 

63 self._ctx = ctx 

64 self.exclude_paths = set( 

65 exclude_paths or [const.DEFAULT_HEALTH_PATH, "/healthz", "/ready"] 

66 ) 

67 

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

69 """ASGI middleware implementation.""" 

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

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

72 return 

73 

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

75 if path in self.exclude_paths: 

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

77 return 

78 

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

80 start_time = ambient_clock.monotonic() 

81 status_code = 500 # Default if response never starts 

82 

83 async def send_with_logging(message: MutableMapping[str, Any]) -> None: 

84 nonlocal status_code 

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

86 status_code = cast("int", message.get("status", 500)) 

87 await send(message) 

88 

89 try: 

90 await self.app(scope, receive, send_with_logging) 

91 finally: 

92 duration_ms = round( 

93 (ambient_clock.monotonic() - start_time) * 1000, 

94 2, 

95 ) 

96 request_id = self._ctx.get(REQUEST_ID) if self._ctx else None 

97 

98 log_kwargs: dict[str, Any] = { 

99 "method": method, 

100 "path": path, 

101 "status": status_code, 

102 "duration_ms": duration_ms, 

103 } 

104 if request_id: 

105 log_kwargs["request_id"] = request_id 

106 

107 if status_code >= 500: 

108 logger.error("request_completed", **log_kwargs) 

109 elif status_code >= 400: 

110 logger.warning("request_completed", **log_kwargs) 

111 else: 

112 logger.info("request_completed", **log_kwargs) 

113 

114 

115__all__ = ["AccessLogMiddleware"]