Coverage for backend/logging_config.py: 18%

1# Adapted from https://gist.github.com/nymous/f138c7f06062b7c43c060bf03759c29e 

2 

3import logging 

4import sys 

5 

6import structlog 

7from structlog.types import EventDict, Processor 

8 

9 

10def _drop_color_message_key(_, __, event_dict: EventDict) -> EventDict: # noqa: ANN001 

11 """ 

12 Remove "color_message" key from the event dict. 

13 

14 Uvicorn logs the message a second time in the extra `color_message`, but we 

15 don't need it. This processor drops the key from the event dict if it 

16 exists. 

17 """ 

18 event_dict.pop("color_message", None) 

19 return event_dict 

20 

21 

22def setup_logging(json_logs: bool = False, log_level: str = "INFO") -> None: 

23 """Setup logging.""" 

24 timestamper = structlog.processors.TimeStamper(fmt="iso") 

25 

26 shared_processors: list[Processor] = [ 

27 structlog.contextvars.merge_contextvars, 

28 structlog.stdlib.add_logger_name, 

29 structlog.stdlib.add_log_level, 

30 structlog.stdlib.PositionalArgumentsFormatter(), 

31 structlog.stdlib.ExtraAdder(), 

32 _drop_color_message_key, 

33 timestamper, 

34 structlog.processors.StackInfoRenderer(), 

35 ] 

36 

37 if json_logs: 

38 # Format the exception only for JSON logs, as we want to pretty-print 

39 # them when using the ConsoleRenderer 

40 shared_processors.append(structlog.processors.format_exc_info) 

41 

42 structlog.configure( 

43 processors=shared_processors 

44 + [ 

45 # Prepare event dict for `ProcessorFormatter`. 

46 structlog.stdlib.ProcessorFormatter.wrap_for_formatter, 

47 ], 

48 logger_factory=structlog.stdlib.LoggerFactory(), 

49 cache_logger_on_first_use=True, 

50 ) 

51 

52 log_renderer: structlog.types.Processor 

53 if json_logs: 

54 log_renderer = structlog.processors.JSONRenderer() 

55 else: 

56 log_renderer = structlog.dev.ConsoleRenderer() 

57 

58 formatter = structlog.stdlib.ProcessorFormatter( 

59 # These run ONLY on `logging` entries that do NOT originate within 

60 # structlog. 

61 foreign_pre_chain=shared_processors, 

62 # These run on ALL entries after the pre_chain is done. 

63 processors=[ 

64 # Remove _record & _from_structlog. 

65 structlog.stdlib.ProcessorFormatter.remove_processors_meta, 

66 log_renderer, 

67 ], 

68 ) 

69 

70 handler = logging.StreamHandler() 

71 # Use OUR `ProcessorFormatter` to format all `logging` entries. 

72 handler.setFormatter(formatter) 

73 root_logger = logging.getLogger() 

74 root_logger.addHandler(handler) 

75 

76 root_logger.setLevel(log_level.upper()) 

77 

78 for _log in ["uvicorn", "uvicorn.error"]: 

79 # Clear the log handlers for uvicorn loggers, and enable propagation so 

80 # the messages are caught by our root logger and formatted correctly by 

81 # structlog 

82 logging.getLogger(_log).handlers.clear() 

83 logging.getLogger(_log).propagate = True 

84 

85 # Since we re-create the access logs ourselves, to add all information in 

86 # the structured log (see the `logging_middleware` in main.py), we clear 

87 # the handlers and prevent the logs to propagate to a logger higher up in 

88 # the hierarchy (effectively rendering them silent). 

89 logging.getLogger("uvicorn.access").handlers.clear() 

90 logging.getLogger("uvicorn.access").propagate = False 

91 

92 def _handle_exception(exc_type, exc_value, exc_traceback) -> None: # noqa: ANN001 

93 """ 

94 Log any uncaught exception instead of letting it be printed by Python. 

95 

96 (But leave KeyboardInterrupt untouched to allow users to Ctrl+C to 

97 stop). 

98 

99 See https://stackoverflow.com/a/16993115/3641865 

100 """ 

101 if issubclass(exc_type, KeyboardInterrupt): 

102 sys.__excepthook__(exc_type, exc_value, exc_traceback) 

103 return 

104 

105 root_logger.error("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback)) 

106 

107 sys.excepthook = _handle_exception