Coverage for polar/logging.py: 88%

1import logging.config 2 ctx1ab

2import uuid 2 ctx1ab

3from typing import Any 2 ctx1ab

4 

5import structlog 2 ctx1ab

6from logfire.integrations.structlog import LogfireProcessor 2 ctx1ab

7 

8from polar.config import settings 2 ctx1ab

9 

10Logger = structlog.stdlib.BoundLogger 2 ctx1ab

11 

12 

13class Logging[RendererType]: 2 ctx1ab

14 """Hubben logging configurator of `structlog` and `logging`. 

15 

16 Customized implementation inspired by the following documentation: 

17 https://www.structlog.org/en/stable/standard-library.html#rendering-using-structlog-based-formatters-within-logging 

18 

19 """ 

20 

21 timestamper = structlog.processors.TimeStamper(fmt="iso") 2 ctx1ab

22 

23 @classmethod 2 ctx1ab

24 def get_level(cls) -> str: 2 ctx1ab

25 return settings.LOG_LEVEL (empty)

26 

27 @classmethod 2 ctx1ab

28 def get_processors(cls, *, logfire: bool) -> list[Any]: 2 ctx1ab

29 return [ (empty)

30 structlog.contextvars.merge_contextvars, 

31 structlog.stdlib.add_log_level, 

32 structlog.stdlib.add_logger_name, 

33 structlog.stdlib.PositionalArgumentsFormatter(), 

34 cls.timestamper, 

35 structlog.processors.UnicodeDecoder(), 

36 structlog.processors.StackInfoRenderer(), 

37 *([LogfireProcessor()] if logfire else []), 

38 structlog.stdlib.ProcessorFormatter.wrap_for_formatter, 

39 ] 

40 

41 @classmethod 2 ctx1ab

42 def get_renderer(cls) -> RendererType: 2 ctx1ab

43 raise NotImplementedError() 

44 

45 @classmethod 2 ctx1ab

46 def configure_stdlib(cls, *, logfire: bool) -> None: 2 ctx1ab

47 level = cls.get_level() (empty)

48 logging.config.dictConfig( (empty)

49 { 

50 "version": 1, 

51 "disable_existing_loggers": True, 

52 "formatters": { 

53 "polar": { 

54 "()": structlog.stdlib.ProcessorFormatter, 

55 "processors": [ 

56 structlog.stdlib.ProcessorFormatter.remove_processors_meta, 

57 cls.get_renderer(), 

58 ], 

59 "foreign_pre_chain": [ 

60 structlog.contextvars.merge_contextvars, 

61 structlog.stdlib.add_log_level, 

62 structlog.stdlib.add_logger_name, 

63 structlog.stdlib.PositionalArgumentsFormatter(), 

64 structlog.stdlib.ExtraAdder(), 

65 cls.timestamper, 

66 structlog.processors.UnicodeDecoder(), 

67 structlog.processors.StackInfoRenderer(), 

68 *([LogfireProcessor()] if logfire else []), 

69 ], 

70 }, 

71 }, 

72 "handlers": { 

73 "default": { 

74 "level": level, 

75 "class": "logging.StreamHandler", 

76 "formatter": "polar", 

77 }, 

78 }, 

79 "loggers": { 

80 "": { 

81 "handlers": ["default"], 

82 "level": level, 

83 "propagate": False, 

84 }, 

85 # Propagate third-party loggers to the root one 

86 **{ 

87 logger: { 

88 "handlers": [], 

89 "propagate": True, 

90 } 

91 for logger in [ 

92 "uvicorn", 

93 "sqlalchemy", 

94 "dramatiq", 

95 "authlib", 

96 "logfire", 

97 "apscheduler", 

98 ] 

99 }, 

100 }, 

101 } 

102 ) 

103 

104 @classmethod 2 ctx1ab

105 def configure_structlog(cls, *, logfire: bool = False) -> None: 2 ctx1ab

106 structlog.configure_once( (empty)

107 processors=cls.get_processors(logfire=logfire), 

108 logger_factory=structlog.stdlib.LoggerFactory(), 

109 wrapper_class=structlog.stdlib.BoundLogger, 

110 cache_logger_on_first_use=True, 

111 ) 

112 

113 @classmethod 2 ctx1ab

114 def configure(cls, *, logfire: bool = False) -> None: 2 ctx1ab

115 cls.configure_stdlib(logfire=logfire) (empty)

116 cls.configure_structlog(logfire=logfire) (empty)

117 

118 

119class Development(Logging[structlog.dev.ConsoleRenderer]): 2 ctx1ab

120 @classmethod 2 ctx1ab

121 def get_renderer(cls) -> structlog.dev.ConsoleRenderer: 2 ctx1ab

122 return structlog.dev.ConsoleRenderer(colors=True) (empty)

123 

124 

125class Production(Logging[structlog.processors.JSONRenderer]): 2 ctx1ab

126 @classmethod 2 ctx1ab

127 def get_renderer(cls) -> structlog.processors.JSONRenderer: 2 ctx1ab

128 return structlog.processors.JSONRenderer() 

129 

130 

131def configure(*, logfire: bool = False) -> None: 2 ctx1ab

132 if settings.is_testing(): 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true(empty)

133 Development.configure(logfire=False) 

134 elif settings.is_development(): 134 ↛ 137line 134 didn't jump to line 137 because the condition on line 134 was always true(empty)

135 Development.configure(logfire=logfire) (empty)

136 else: 

137 Production.configure(logfire=logfire) 

138 

139 

140def generate_correlation_id() -> str: 2 ctx1ab

141 return str(uuid.uuid4()) 2 ctx1dc