Coverage for src/qdrant_loader_core/logging.py: 88%
129 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:12 +0000
1"""Unified logging configuration for qdrant-loader ecosystem.
3Provides:
4- structlog setup (console/json/file) with redaction
5- stdlib logging bridge with redaction filter
6- optional suppression of noisy third-party logs
7"""
9from __future__ import annotations
11import logging
12import os
14import structlog
15from structlog.stdlib import LoggerFactory
17from .logging_filters import (
18 ApplicationFilter,
19 QdrantVersionFilter,
20 RedactionFilter,
21 UvicornAccessRedactFilter,
22)
23from .logging_processors import CleanFormatter, redact_processor
25try:
26 # ExtraAdder is available in structlog >= 20
27 from structlog.stdlib import ExtraAdder # type: ignore
28except Exception: # pragma: no cover - fallback when absent
29 ExtraAdder = None # type: ignore
32class LoggingConfig:
33 """Core logging setup with structlog + stdlib redaction and filters."""
35 _initialized = False
36 _installed_handlers: list[logging.Handler] = []
37 _file_handler: logging.FileHandler | None = None
38 _current_config: (
39 tuple[
40 str, # level
41 str, # format
42 str | None, # file
43 bool, # clean_output
44 bool, # suppress_qdrant_warnings
45 bool, # disable_console
46 ]
47 | None
48 ) = None
50 @classmethod
51 def setup(
52 cls,
53 *,
54 level: str = "INFO",
55 format: str = "console", # "console" | "json"
56 file: str | None = None,
57 clean_output: bool = True,
58 suppress_qdrant_warnings: bool = True,
59 disable_console: bool | None = None,
60 ) -> None:
61 # Env override for console toggling (e.g., MCP server)
62 if disable_console is None:
63 disable_console = (
64 os.getenv("MCP_DISABLE_CONSOLE_LOGGING", "").lower() == "true"
65 )
67 try:
68 numeric_level = getattr(logging, level.upper())
69 except AttributeError:
70 raise ValueError(f"Invalid log level: {level}") from None
72 # Short-circuit when configuration is unchanged
73 current_tuple = (
74 level.upper(),
75 format,
76 file,
77 bool(clean_output),
78 bool(suppress_qdrant_warnings),
79 bool(disable_console),
80 )
81 if cls._initialized and cls._current_config == current_tuple:
82 return
84 # Reset structlog defaults but preserve existing stdlib handlers (e.g., pytest caplog)
85 structlog.reset_defaults()
87 # Remove any handlers previously added by this class, and also clear
88 # any pre-existing root handlers that may cause duplicated outputs.
89 # We keep this conservative by only touching the root logger.
90 root_logger = logging.getLogger()
91 # First remove our previously installed handlers
92 for h in list(cls._installed_handlers):
93 try:
94 root_logger.removeHandler(h)
95 if isinstance(h, logging.FileHandler):
96 try:
97 h.close()
98 except Exception:
99 pass
100 except Exception:
101 pass
102 cls._installed_handlers.clear()
104 # Then remove any remaining handlers on the root logger (e.g., added by
105 # earlier setup calls or third-parties) to avoid duplicate emissions.
106 # This is safe for CLI usage; tests relying on caplog attach to non-root loggers.
107 for h in list(root_logger.handlers):
108 try:
109 root_logger.removeHandler(h)
110 if isinstance(h, logging.FileHandler):
111 try:
112 h.close()
113 except Exception:
114 pass
115 except Exception:
116 pass
118 handlers: list[logging.Handler] = []
120 # Choose timestamp format and final renderer for structlog messages
121 if clean_output and format == "console":
122 ts_fmt = "%H:%M:%S"
123 final_renderer = structlog.dev.ConsoleRenderer(colors=True)
124 else:
125 ts_fmt = "iso"
126 final_renderer = (
127 structlog.processors.JSONRenderer()
128 if format == "json"
129 else structlog.dev.ConsoleRenderer(colors=True)
130 )
132 if not disable_console:
133 console_handler = logging.StreamHandler()
134 console_handler.setFormatter(logging.Formatter("%(message)s"))
135 console_handler.addFilter(ApplicationFilter())
136 console_handler.addFilter(RedactionFilter())
137 handlers.append(console_handler)
139 if file:
140 file_handler = logging.FileHandler(file)
141 # Use CleanFormatter to strip ANSI sequences from structlog console renderer output
142 file_handler.setFormatter(CleanFormatter("%(message)s"))
143 file_handler.addFilter(ApplicationFilter())
144 file_handler.addFilter(RedactionFilter())
145 handlers.append(file_handler)
147 # Attach our handlers without removing existing ones (so pytest caplog keeps working)
148 root_logger.setLevel(numeric_level)
149 for h in handlers:
150 root_logger.addHandler(h)
151 # Track handlers we installed to avoid duplicates on re-setup
152 cls._installed_handlers.extend(handlers)
153 # Track file handler for lightweight reconfiguration
154 cls._file_handler = next(
155 (h for h in handlers if isinstance(h, logging.FileHandler)), None
156 )
158 # Add global filters so captured logs (e.g., pytest caplog) are also redacted
159 # Avoid duplicate filters if setup() is called multiple times
160 has_redaction = any(isinstance(f, RedactionFilter) for f in root_logger.filters)
161 if not has_redaction:
162 root_logger.addFilter(RedactionFilter())
163 has_app_filter = any(
164 isinstance(f, ApplicationFilter) for f in root_logger.filters
165 )
166 if not has_app_filter:
167 root_logger.addFilter(ApplicationFilter())
169 # uvicorn.access has propagate=False in uvicorn's own logging config, so
170 # it bypasses the root logger's RedactionFilter above. Attach directly
171 # so webhook secrets/tokens in the request query string aren't logged
172 # in plaintext (see UvicornAccessRedactFilter docstring for details).
173 uvicorn_access_logger = logging.getLogger("uvicorn.access")
174 if not any(
175 isinstance(f, UvicornAccessRedactFilter)
176 for f in uvicorn_access_logger.filters
177 ):
178 uvicorn_access_logger.addFilter(UvicornAccessRedactFilter())
180 # Optional suppressions
181 if suppress_qdrant_warnings:
182 logging.getLogger("qdrant_client").addFilter(QdrantVersionFilter())
184 # Quiet noisy libs a bit
185 for name in ("httpx", "httpcore", "urllib3", "gensim"):
186 logging.getLogger(name).setLevel(logging.WARNING)
188 # structlog processors – render to a final string directly
189 structlog.configure(
190 processors=[
191 structlog.stdlib.filter_by_level,
192 structlog.stdlib.add_logger_name,
193 structlog.stdlib.add_log_level,
194 structlog.processors.TimeStamper(fmt=ts_fmt),
195 redact_processor,
196 final_renderer,
197 ],
198 wrapper_class=structlog.make_filtering_bound_logger(numeric_level),
199 logger_factory=LoggerFactory(),
200 cache_logger_on_first_use=False,
201 )
203 cls._initialized = True
204 cls._current_config = current_tuple
206 @classmethod
207 def get_logger(cls, name: str | None = None) -> structlog.BoundLogger:
208 if not cls._initialized:
209 cls.setup()
210 return structlog.get_logger(name)
212 @classmethod
213 def reconfigure(cls, *, file: str | None = None, level: str | None = None) -> None:
214 """Lightweight reconfiguration for file destination and optionally log level.
216 Replaces only the file handler while keeping console handlers and
217 structlog processors intact. Optionally updates the log level.
219 Args:
220 file: Path to log file (optional)
221 level: New log level (optional, e.g., "DEBUG", "INFO")
222 """
223 root_logger = logging.getLogger()
225 # Update log level if provided
226 if level is not None:
227 try:
228 numeric_level = getattr(logging, level.upper())
229 root_logger.setLevel(numeric_level)
231 # Update structlog wrapper to use new level
232 if cls._current_config is not None:
233 (
234 _,
235 fmt,
236 _,
237 clean_output,
238 suppress_qdrant_warnings,
239 disable_console,
240 ) = cls._current_config
242 # Choose timestamp format and final renderer
243 if clean_output and fmt == "console":
244 ts_fmt = "%H:%M:%S"
245 final_renderer = structlog.dev.ConsoleRenderer(colors=True)
246 else:
247 ts_fmt = "iso"
248 final_renderer = (
249 structlog.processors.JSONRenderer()
250 if fmt == "json"
251 else structlog.dev.ConsoleRenderer(colors=True)
252 )
254 # Reconfigure structlog with new level
255 structlog.configure(
256 processors=[
257 structlog.stdlib.filter_by_level,
258 structlog.stdlib.add_logger_name,
259 structlog.stdlib.add_log_level,
260 structlog.processors.TimeStamper(fmt=ts_fmt),
261 redact_processor,
262 final_renderer,
263 ],
264 wrapper_class=structlog.make_filtering_bound_logger(
265 numeric_level
266 ),
267 logger_factory=LoggerFactory(),
268 cache_logger_on_first_use=False,
269 )
270 except AttributeError:
271 raise ValueError(f"Invalid log level: {level}") from None
273 # Remove existing file handler if present
274 if cls._file_handler is not None:
275 try:
276 root_logger.removeHandler(cls._file_handler)
277 cls._file_handler.close()
278 except Exception:
279 pass
280 cls._installed_handlers = [
281 h for h in cls._installed_handlers if h is not cls._file_handler
282 ]
283 cls._file_handler = None
285 # Add new file handler if requested
286 if file:
287 fh = logging.FileHandler(file)
288 fh.setFormatter(CleanFormatter("%(message)s"))
289 fh.addFilter(ApplicationFilter())
290 fh.addFilter(RedactionFilter())
291 root_logger.addHandler(fh)
292 cls._installed_handlers.append(fh)
293 cls._file_handler = fh
295 # Update current config tuple if available
296 if cls._current_config is not None:
297 (
298 old_level,
299 fmt,
300 _,
301 clean_output,
302 suppress_qdrant_warnings,
303 disable_console,
304 ) = cls._current_config
305 new_level = level.upper() if level is not None else old_level
306 cls._current_config = (
307 new_level,
308 fmt,
309 file,
310 clean_output,
311 suppress_qdrant_warnings,
312 disable_console,
313 )