Coverage for logging_config.py: 100.00%

1""" 

2Structured Logging Configuration 

3 

4Provides consistent structured JSON logging for backend and Celery workers, 

5optimized for Grafana Loki integration. 

6 

7Usage: 

8 from logging_config import setup_logging 

9 

10 setup_logging(service_name="backend") 

11 logger = structlog.get_logger() 

12 logger.info("event_occurred", user_id=123, action="login") 

13""" 

14 

15import logging 

16import os 

17from typing import Any 

18 

19import structlog 

20 

21 

22def setup_logging(service_name: str = "missing-table") -> None: 

23 """ 

24 Configure structured logging with JSON output for Loki/Grafana. 

25 

26 Args: 

27 service_name: Name of the service (backend, celery-worker, etc.) 

28 """ 

29 log_level = os.getenv("LOG_LEVEL", "INFO").upper() 

30 

31 # Configure standard logging 

32 logging.basicConfig( 

33 format="%(message)s", 

34 level=getattr(logging, log_level, logging.INFO), 

35 handlers=[logging.StreamHandler()], 

36 ) 

37 

38 # Suppress verbose httpcore/httpx debug logs (they're too noisy) 

39 # Only show warnings and errors from httpcore/httpx 

40 logging.getLogger("httpcore").setLevel(logging.WARNING) 

41 logging.getLogger("httpx").setLevel(logging.WARNING) 

42 

43 # Configure structlog with JSON renderer for Loki 

44 structlog.configure( 

45 processors=[ 

46 # Merge context variables (session_id, request_id from middleware) 

47 structlog.contextvars.merge_contextvars, 

48 # Add log level to event dict 

49 structlog.stdlib.add_log_level, 

50 # Add logger name to event dict 

51 structlog.stdlib.add_logger_name, 

52 # Add timestamp in ISO format 

53 structlog.processors.TimeStamper(fmt="iso"), 

54 # Add stack info for exceptions 

55 structlog.processors.StackInfoRenderer(), 

56 # Format exceptions nicely 

57 structlog.processors.format_exc_info, 

58 # Decode unicode 

59 structlog.processors.UnicodeDecoder(), 

60 # Add callsite info (filename, line number) 

61 structlog.processors.CallsiteParameterAdder( 

62 parameters=[ 

63 structlog.processors.CallsiteParameter.FILENAME, 

64 structlog.processors.CallsiteParameter.LINENO, 

65 ] 

66 ), 

67 # Render as JSON for Loki 

68 structlog.processors.JSONRenderer(), 

69 ], 

70 # Use logging module as backend 

71 wrapper_class=structlog.stdlib.BoundLogger, 

72 context_class=dict, 

73 logger_factory=structlog.stdlib.LoggerFactory(), 

74 cache_logger_on_first_use=True, 

75 ) 

76 

77 # Bind service name globally 

78 structlog.contextvars.bind_contextvars(service=service_name) 

79 

80 # Log initialization 

81 logger = structlog.get_logger() 

82 logger.info("logging_initialized", service=service_name, log_level=log_level, format="json") 

83 

84 

85def get_logger(name: str | None = None) -> Any: 

86 """ 

87 Get a structlog logger instance. 

88 

89 Args: 

90 name: Optional logger name (typically __name__) 

91 

92 Returns: 

93 Configured structlog logger 

94 

95 Example: 

96 logger = get_logger(__name__) 

97 logger.info("user_login", user_id=123, method="oauth") 

98 """ 

99 return structlog.get_logger(name)