Coverage for polar/logging.py: 88%
45 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1import logging.config
2import uuid
3from typing import Any
5import structlog
6from logfire.integrations.structlog import LogfireProcessor
8from polar.config import settings
10Logger = structlog.stdlib.BoundLogger
13class Logging[RendererType]:
14 """Hubben logging configurator of `structlog` and `logging`.
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
19 """
21 timestamper = structlog.processors.TimeStamper(fmt="iso")
23 @classmethod
24 def get_level(cls) -> str:
25 return settings.LOG_LEVEL
27 @classmethod
28 def get_processors(cls, *, logfire: bool) -> list[Any]:
29 return [
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 ]
41 @classmethod
42 def get_renderer(cls) -> RendererType:
43 raise NotImplementedError()
45 @classmethod
46 def configure_stdlib(cls, *, logfire: bool) -> None:
47 level = cls.get_level()
48 logging.config.dictConfig(
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 )
104 @classmethod
105 def configure_structlog(cls, *, logfire: bool = False) -> None:
106 structlog.configure_once(
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 )
113 @classmethod
114 def configure(cls, *, logfire: bool = False) -> None:
115 cls.configure_stdlib(logfire=logfire)
116 cls.configure_structlog(logfire=logfire)
119class Development(Logging[structlog.dev.ConsoleRenderer]):
120 @classmethod
121 def get_renderer(cls) -> structlog.dev.ConsoleRenderer:
122 return structlog.dev.ConsoleRenderer(colors=True)
125class Production(Logging[structlog.processors.JSONRenderer]):
126 @classmethod
127 def get_renderer(cls) -> structlog.processors.JSONRenderer:
128 return structlog.processors.JSONRenderer()
131def configure(*, logfire: bool = False) -> None:
132 if settings.is_testing(): 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true
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
135 Development.configure(logfire=logfire)
136 else:
137 Production.configure(logfire=logfire)
140def generate_correlation_id() -> str:
141 return str(uuid.uuid4())