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