Coverage for app/core/logging.py: 95%

1import logging 

2import sys 

3from typing import Any 

4import structlog 

5from structlog.typing import EventDict 

6 

7from app.core.context import get_correlation_id 

8 

9 

10def _add_correlation_id(_: Any, __: str, event_dict: EventDict) -> EventDict: 

11 event_dict["correlation_id"] = get_correlation_id() 

12 return event_dict 

13 

14 

15def logging_config() -> None: 

16 logging.basicConfig( 

17 format="%(message)s", 

18 stream=sys.stdout, 

19 level=logging.INFO, 

20 ) 

21 

22 structlog.configure( 

23 processors=[ 

24 structlog.contextvars.merge_contextvars, 

25 _add_correlation_id, 

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

27 structlog.processors.add_log_level, 

28 structlog.processors.JSONRenderer(), 

29 ], 

30 logger_factory=structlog.PrintLoggerFactory(), 

31 ) 

32 

33 

34class Logger: 

35 _logger = structlog.get_logger() 

36 

37 @classmethod 

38 def log(cls, event: str, **kwargs: Any) -> None: 

39 cls._logger.info(event.upper(), **kwargs) 

40 

41 @classmethod 

42 def error(cls, event: str, **kwargs: Any) -> None: 

43 cls._logger.error(event.upper(), **kwargs)