Coverage for src/lumen/logging.py: 100%

1""" 

2This file sets up the logging configuration using structlog. 

3It integrates JSON formatting and request correlation IDs for structured backend logs. 

4""" 

5 

6import logging 

7 

8import structlog 

9from asgi_correlation_id import correlation_id 

10from structlog.typing import EventDict 

11 

12from .observability.log_context import add_otel_trace_context 

13 

14 

15def add_correlation_id( 

16 logger: logging.Logger, method_name: str, event_dict: EventDict 

17) -> EventDict: 

18 if req_id := correlation_id.get(): 

19 event_dict["request_id"] = req_id 

20 return event_dict 

21 

22 

23def configure_logging(level: str = "info", fmt: str = "console") -> None: 

24 shared_processors: list[structlog.types.Processor] = [ 

25 structlog.stdlib.filter_by_level, 

26 structlog.contextvars.merge_contextvars, 

27 add_correlation_id, 

28 add_otel_trace_context, 

29 structlog.stdlib.add_logger_name, 

30 structlog.stdlib.add_log_level, 

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

32 structlog.processors.StackInfoRenderer(), 

33 ] 

34 

35 renderer: structlog.types.Processor 

36 if fmt == "json": 

37 renderer = structlog.processors.JSONRenderer() 

38 else: 

39 renderer = structlog.dev.ConsoleRenderer() 

40 

41 structlog.configure( 

42 processors=[*shared_processors, renderer], 

43 wrapper_class=structlog.stdlib.BoundLogger, 

44 context_class=dict, 

45 logger_factory=structlog.stdlib.LoggerFactory(), 

46 cache_logger_on_first_use=True, 

47 ) 

48 # `logging.basicConfig()` is a no-op once the root logger already has a handler — which 

49 # happens whenever `setup_telemetry()` ran first and attached the OTel log-export bridge 

50 # handler (LUMEN_OTEL_LOG_EXPORT=true; see observability/otel.py::_setup_log_bridge). Left 

51 # to basicConfig alone, the root logger silently stays at its Python default (WARNING), 

52 # dropping every `.info()` call — including the entire per-request access log — with no 

53 # error of any kind. Set the level explicitly, and always ensure our own stdout handler 

54 # exists (checked by marker attribute, not "any handler") so stdout-JSON (Weg A) stays the 

55 # default even when the OTel bridge handler was already attached first. 

56 root_logger = logging.getLogger() 

57 root_logger.setLevel(getattr(logging, level.upper())) 

58 if not any(getattr(h, "_lumen_stdout_handler", False) for h in root_logger.handlers): 

59 stream_handler = logging.StreamHandler() 

60 stream_handler.setFormatter(logging.Formatter("%(message)s")) 

61 stream_handler._lumen_stdout_handler = True # type: ignore[attr-defined] 

62 root_logger.addHandler(stream_handler)