Coverage for src/lumen/core/logging_middleware.py: 83%

1""" 

2This middleware handles access logging for all HTTP requests. 

3It extracts details like timing and client IP to output structured logs. 

4""" 

5 

6import time 

7from collections.abc import Awaitable, Callable 

8 

9import structlog 

10from fastapi import Request, Response 

11from starlette.middleware.base import BaseHTTPMiddleware 

12 

13from .context import current_client_ip 

14 

15logger = structlog.get_logger("access") 

16 

17 

18class AccessLogMiddleware(BaseHTTPMiddleware): 

19 """ 

20 - Middleware to log access requests and responses 

21 - It logs the HTTP method, path, status code, duration, and client IP address 

22 - checks requests and responses for exceptions and logs them accordingly 

23 """ 

24 

25 async def dispatch( 

26 self, request: Request, call_next: Callable[[Request], Awaitable[Response]] 

27 ) -> Response: 

28 start_time = time.perf_counter() 

29 structlog.contextvars.clear_contextvars() 

30 

31 client_host = request.client.host if request.client else None 

32 ip_token = current_client_ip.set(client_host) 

33 

34 try: 

35 try: 

36 response = await call_next(request) 

37 process_time = time.perf_counter() - start_time 

38 

39 logger.info( 

40 "http.request", 

41 method=request.method, 

42 path=request.url.path, 

43 status_code=response.status_code, 

44 duration_ms=round(process_time * 1000, 2), 

45 client_ip=request.client.host if request.client else None, 

46 ) 

47 return response 

48 except Exception as e: 

49 process_time = time.perf_counter() - start_time 

50 logger.error( 

51 "http.request.error", 

52 method=request.method, 

53 path=request.url.path, 

54 duration_ms=round(process_time * 1000, 2), 

55 client_ip=request.client.host if request.client else None, 

56 exc_info=e, 

57 ) 

58 raise 

59 finally: 

60 current_client_ip.reset(ip_token)