Coverage for src/qdrant_loader/utils/logging.py: 38%
269 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""Centralized logging configuration for the application."""
3import logging
4import re
5from urllib.parse import unquote_plus
7import structlog
9from qdrant_loader.utils.sensitive import redact_sensitive_data
12class QdrantVersionFilter(logging.Filter):
13 """Filter to suppress Qdrant version check warnings."""
15 def filter(self, record):
16 return "version check" not in record.getMessage().lower()
19class ApplicationFilter(logging.Filter):
20 """Filter to only show logs from our application."""
22 def filter(self, record):
23 # Only show logs from our application
24 return not record.name.startswith(("httpx", "httpcore", "urllib3"))
27class SQLiteFilter(logging.Filter):
28 """Filter to suppress verbose SQLite operation logs."""
30 def filter(self, record):
31 # Suppress verbose SQLite debug logs
32 message = record.getMessage()
33 sqlite_patterns = [
34 "executing functools.partial(<built-in method",
35 "operation functools.partial(<built-in method",
36 "executing <function connect.",
37 "operation <function connect.",
38 "returning exception",
39 ]
40 return not any(pattern in message for pattern in sqlite_patterns)
43class VerbosityFilter(logging.Filter):
44 """Filter to reduce verbosity of debug messages."""
46 def filter(self, record):
47 # Only filter DEBUG level messages
48 if record.levelno != logging.DEBUG:
49 return True
51 # Suppress overly verbose debug messages
52 message = record.getMessage()
53 verbose_patterns = [
54 "HTTP Request:",
55 "Response status:",
56 "Request headers:",
57 "Response headers:",
58 # PDF parsing debug messages
59 "seek:",
60 "nexttoken:",
61 "do_keyword:",
62 "nextobject:",
63 "add_results:",
64 "register:",
65 "getobj:",
66 "get_unichr:",
67 "exec:",
68 # Character encoding detection
69 "confidence =",
70 "prober hit error",
71 "not active",
72 # File processing verbosity
73 "checking if file should be processed",
74 "current configuration",
75 "checking file extension",
76 "file type detection",
77 "file supported via",
78 "starting metadata extraction",
79 "completed metadata extraction",
80 "document metadata:",
81 # HTTP client debug
82 "connect_tcp.started",
83 "connect_tcp.complete",
84 "send_request_headers",
85 "send_request_body",
86 "receive_response_headers",
87 "receive_response_body",
88 "response_closed",
89 ]
90 return not any(pattern in message for pattern in verbose_patterns)
93class UvicornAccessRedactFilter(logging.Filter):
94 """Redacts secret/token query-string values from uvicorn's access log line.
96 uvicorn logs the raw request line (including the query string) via
97 ``record.args`` on the "uvicorn.access" logger, e.g.
98 ``args = (client_addr, method, "/path?token=abc123", http_version, status)``.
99 That logger has ``propagate=False`` in uvicorn's default logging config, so
100 it never reaches the root logger's handlers/filters — the secret would
101 otherwise be written to the access log in plaintext. Attach this filter
102 directly to the "uvicorn.access" logger; per-logger filters run in
103 ``Logger.handle()`` before any handler, so this applies regardless of what
104 handlers uvicorn's own dictConfig installs (dictConfig only replaces
105 handlers, not filters).
106 """
108 _SENSITIVE_QUERY_KEYS = {
109 "token",
110 "secret",
111 "signature",
112 "password",
113 "authorization",
114 "api_key",
115 "api-key",
116 "access_key",
117 "access-key",
118 "private_key",
119 "private-key",
120 "access_token",
121 "access-token",
122 }
123 _QUERY_PAIR = re.compile(r'([?&])([^=&\s"]+)=([^&\s"]*)')
125 @classmethod
126 def _redact_query_pair(cls, match: re.Match[str]) -> str:
127 sep, raw_key, _value = match.groups()
128 # uvicorn logs the raw, still percent-encoded request line, so a key
129 # like "sec%72et" would slip past a literal match while Starlette/
130 # FastAPI (which percent-decodes query keys during parsing) still
131 # resolves it to "secret" and accepts it as the webhook secret.
132 # Decode before comparing so encoded keys are caught too.
133 if unquote_plus(raw_key).lower() in cls._SENSITIVE_QUERY_KEYS:
134 return f"{sep}{raw_key}=***REDACTED***"
135 return match.group(0)
137 def filter(self, record: logging.LogRecord) -> bool:
138 try:
139 if isinstance(record.args, tuple) and len(record.args) >= 3:
140 path = record.args[2]
141 if isinstance(path, str) and "?" in path:
142 redacted = self._QUERY_PAIR.sub(self._redact_query_pair, path)
143 if redacted != path:
144 record.args = (
145 record.args[0],
146 record.args[1],
147 redacted,
148 *record.args[3:],
149 )
150 except Exception:
151 pass
152 return True
155class WindowsSafeConsoleHandler(logging.StreamHandler):
156 """Custom console handler that handles Windows encoding issues."""
158 # ANSI escape sequence pattern
159 ANSI_ESCAPE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
161 def emit(self, record):
162 """Emit a record, handling Windows console encoding issues."""
163 try:
164 # Check if stream is still open before writing
165 if hasattr(self.stream, "closed") and self.stream.closed:
166 return # Skip logging if stream is closed
168 # Get the formatted message
169 msg = self.format(record)
171 # For Windows, handle Unicode characters proactively
172 import sys
174 if sys.platform == "win32":
175 # Replace problematic Unicode characters with safe alternatives
176 safe_msg = (
177 msg.replace("🚀", "[ROCKET]")
178 .replace("📄", "[DOCUMENT]")
179 .replace("✅", "[CHECK]")
180 .replace("❌", "[CROSS]")
181 .replace("⚠️", "[WARNING]")
182 .replace("🔍", "[SEARCH]")
183 .replace("💾", "[SAVE]")
184 .replace("🔧", "[TOOL]")
185 .replace("📊", "[CHART]")
186 .replace("🎯", "[TARGET]")
187 .replace("⚙️", "[GEAR]")
188 .replace("⏱️", "[TIMER]")
189 .replace("⏭️", "[NEXT]")
190 .replace("🏗️", "[CONSTRUCTION]")
191 .replace("1️⃣", "[1]")
192 .replace("2️⃣", "[2]")
193 .replace("3️⃣", "[3]")
194 .replace("4️⃣", "[4]")
195 )
196 try:
197 self.stream.write(safe_msg + self.terminator)
198 except UnicodeEncodeError:
199 # Final fallback: encode with ASCII and replace all unsupported chars
200 ascii_msg = safe_msg.encode("ascii", errors="replace").decode(
201 "ascii"
202 )
203 self.stream.write(ascii_msg + self.terminator)
204 else:
205 # For non-Windows, use standard handling
206 self.stream.write(msg + self.terminator)
208 self.flush()
209 except (ValueError, OSError):
210 # Stream is closed or unavailable, skip silently
211 pass
212 except Exception:
213 self.handleError(record)
216class CleanFileHandler(logging.FileHandler):
217 """Custom file handler that strips ANSI color codes from log messages."""
219 # ANSI escape sequence pattern
220 ANSI_ESCAPE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
222 def emit(self, record):
223 """Emit a record, stripping ANSI codes from the message."""
224 try:
225 # Check if stream is still open before writing
226 if hasattr(self.stream, "closed") and self.stream.closed:
227 return # Skip logging if stream is closed
229 # Get the formatted message
230 msg = self.format(record)
231 # Strip ANSI escape sequences
232 clean_msg = self.ANSI_ESCAPE.sub("", msg)
234 # Handle Windows console encoding issues more robustly
235 stream = self.stream
236 try:
237 stream.write(clean_msg + self.terminator)
238 except UnicodeEncodeError:
239 # Fallback for Windows console encoding issues
240 # Replace problematic Unicode characters with safe alternatives
241 import sys
243 if sys.platform == "win32":
244 # Replace common emoji/Unicode characters with ASCII equivalents
245 safe_msg = (
246 clean_msg.replace("🚀", "[ROCKET]")
247 .replace("📄", "[DOCUMENT]")
248 .replace("✅", "[CHECK]")
249 .replace("❌", "[CROSS]")
250 .replace("⚠️", "[WARNING]")
251 .replace("🔍", "[SEARCH]")
252 .replace("💾", "[SAVE]")
253 .replace("🔧", "[TOOL]")
254 .replace("📊", "[CHART]")
255 .replace("🎯", "[TARGET]")
256 .replace("⚙️", "[GEAR]")
257 .replace("⏱️", "[TIMER]")
258 .replace("⏭️", "[NEXT]")
259 .replace("🏗️", "[CONSTRUCTION]")
260 .replace("1️⃣", "[1]")
261 .replace("2️⃣", "[2]")
262 .replace("3️⃣", "[3]")
263 .replace("4️⃣", "[4]")
264 )
265 try:
266 stream.write(safe_msg + self.terminator)
267 except UnicodeEncodeError:
268 # Final fallback: encode with ASCII and replace all unsupported chars
269 ascii_msg = safe_msg.encode("ascii", errors="replace").decode(
270 "ascii"
271 )
272 stream.write(ascii_msg + self.terminator)
273 else:
274 # For non-Windows, try UTF-8 encoding with replacement
275 safe_msg = clean_msg.encode("utf-8", errors="replace").decode(
276 "utf-8", errors="replace"
277 )
278 stream.write(safe_msg + self.terminator)
279 self.flush()
280 except (ValueError, OSError):
281 # Stream is closed or unavailable, skip silently
282 pass
283 except Exception:
284 self.handleError(record)
287class CleanFormatter(logging.Formatter):
288 """Custom formatter that shows only the message for INFO level logs."""
290 def __init__(self, use_custom_renderer=False):
291 super().__init__()
292 self.use_custom_renderer = use_custom_renderer
294 def format(self, record):
295 message = record.getMessage()
297 # If we're using the custom renderer, just return the message as-is
298 # since the CustomConsoleRenderer already handled the formatting
299 if self.use_custom_renderer:
300 return message
302 # For INFO level, just show the message
303 if record.levelno == logging.INFO:
304 return message
305 else:
306 # For other levels, we need to reorder timestamp and level
307 # Check if message starts with a timestamp (HH:MM:SS format)
308 # The message might contain ANSI color codes, so we need to account for that
309 time_pattern = (
310 r"^(?:\x1b\[[0-9;]*m)?(\d{2}:\d{2}:\d{2})(?:\x1b\[[0-9;]*m)?\s+(.*)"
311 )
312 match = re.match(time_pattern, message)
314 if match:
315 timestamp = match.group(1)
316 rest_of_message = match.group(2)
318 # Check if the rest_of_message already contains a level tag
319 # This can happen when CustomConsoleRenderer already formatted it
320 level_in_message_pattern = (
321 r"^\[(?:DEBUG|INFO|WARNING|ERROR|CRITICAL)\]\s"
322 )
323 if re.match(level_in_message_pattern, rest_of_message):
324 # Level already present, don't add another one
325 return f"{timestamp} {rest_of_message}"
326 else:
327 # Add level tag
328 return f"{timestamp} [{record.levelname}] {rest_of_message}"
329 else:
330 # No timestamp found
331 # Check if message already has a level tag at the beginning
332 level_in_message_pattern = (
333 r"^\[(?:DEBUG|INFO|WARNING|ERROR|CRITICAL)\]\s"
334 )
335 if re.match(level_in_message_pattern, message):
336 # Level already present, don't add another one
337 return message
338 else:
339 # Add level tag
340 return f"[{record.levelname}] {message}"
343class FileRenderer:
344 """Custom renderer for file output without timestamps (FileFormatter will add them)."""
346 def __call__(self, logger, method_name, event_dict):
347 # Extract the main message
348 event = event_dict.pop("event", "")
350 # Format additional key-value pairs
351 if event_dict:
352 extras = " ".join(f"{k}={v}" for k, v in event_dict.items())
353 return f"{event} {extras}".strip()
354 else:
355 return event
358class FileFormatter(logging.Formatter):
359 """Custom formatter for file output that provides clean, readable logs without ANSI codes."""
361 # ANSI escape sequence pattern
362 ANSI_ESCAPE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
364 def format(self, record):
365 # Get the timestamp
366 timestamp = self.formatTime(record, "%Y-%m-%d %H:%M:%S")
368 # Get the level name
369 level = record.levelname
371 # Get the message (this will be the already formatted structlog message)
372 message = record.getMessage()
374 # First, strip ANSI escape sequences
375 clean_message = self.ANSI_ESCAPE.sub("", message)
377 # Now check if the clean message starts with a timestamp and remove it
378 # Pattern for structlog output: "HH:MM:SS message content"
379 time_pattern = r"^\d{2}:\d{2}:\d{2}\s+"
380 if re.match(time_pattern, clean_message):
381 # Remove the structlog timestamp since we're adding our own
382 clean_message = re.sub(time_pattern, "", clean_message)
384 # Check if the message already contains a level tag to avoid duplication
385 level_in_message_pattern = r"^\[(?:DEBUG|INFO|WARNING|ERROR|CRITICAL)\]\s"
386 has_level_tag = re.match(level_in_message_pattern, clean_message)
388 # Format based on log level
389 if record.levelno == logging.INFO:
390 # For INFO level, use a clean format: timestamp | message
391 return f"{timestamp} | {clean_message}"
392 else:
393 # For other levels, include the level only if not already present
394 if has_level_tag:
395 return f"{timestamp} | {clean_message}"
396 else:
397 return f"{timestamp} | [{level}] {clean_message}"
400class CustomConsoleRenderer:
401 """Custom console renderer that formats timestamp and level correctly."""
403 def __init__(self, colors=True):
404 self.colors = colors
405 self._console_renderer = structlog.dev.ConsoleRenderer(colors=colors)
406 # ANSI color codes
407 self.gray = "\033[90m" if colors else ""
408 self.green = "\033[92m" if colors else "" # Bright green for INFO
409 self.yellow = "\033[93m" if colors else "" # Bright yellow for WARNING
410 self.red = "\033[91m" if colors else "" # Bright red for ERROR
411 self.magenta = "\033[95m" if colors else "" # Bright magenta for CRITICAL
412 self.cyan = "\033[96m" if colors else "" # Bright cyan for DEBUG
413 self.reset = "\033[0m" if colors else ""
415 def _get_level_color(self, level):
416 """Get the appropriate color for a log level."""
417 level_colors = {
418 "DEBUG": self.cyan,
419 "INFO": self.green, # Green for INFO
420 "WARNING": self.yellow,
421 "ERROR": self.red,
422 "CRITICAL": self.magenta,
423 }
424 return level_colors.get(level, "")
426 def __call__(self, logger, method_name, event_dict):
427 # Extract timestamp if present
428 timestamp = event_dict.pop("timestamp", None)
430 # Get the level from method_name
431 level = method_name.upper()
433 # Use the default console renderer to format the rest
434 formatted = self._console_renderer(logger, method_name, event_dict)
436 # If we have a timestamp
437 if timestamp and isinstance(timestamp, str) and len(timestamp) >= 8:
438 time_part = timestamp[:8] # Get HH:MM:SS part
440 # Remove the timestamp from the formatted message if it's there
441 if formatted.startswith(time_part):
442 formatted = formatted[len(time_part) :].lstrip()
444 # Add gray color to timestamp
445 colored_timestamp = f"{self.gray}{time_part}{self.reset}"
447 # Get colored level for all levels including INFO
448 level_color = self._get_level_color(level)
449 colored_level = (
450 f"{level_color}[{level}]{self.reset}" if level_color else f"[{level}]"
451 )
453 # Show timestamp, colored level, and message for all levels
454 return f"{colored_timestamp} {colored_level} {formatted}"
456 # Fallback if no timestamp
457 level_color = self._get_level_color(level)
458 colored_level = (
459 f"{level_color}[{level}]{self.reset}" if level_color else f"[{level}]"
460 )
461 return f"{colored_level} {formatted}"
464class LoggingConfig:
465 """Centralized logging configuration."""
467 _initialized = False
468 _current_config = None
470 @classmethod
471 def setup(
472 cls,
473 level: str = "INFO",
474 format: str = "console",
475 file: str | None = None,
476 suppress_qdrant_warnings: bool = True,
477 clean_output: bool = True,
478 ) -> None:
479 """Setup logging configuration.
481 Args:
482 level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
483 format: Log format (json or console)
484 file: Path to log file (optional)
485 suppress_qdrant_warnings: Whether to suppress Qdrant version check warnings
486 clean_output: Whether to use clean, less verbose output
487 """
488 try:
489 # Convert string level to logging level
490 numeric_level = getattr(logging, level.upper())
491 except AttributeError:
492 raise ValueError(f"Invalid log level: {level}") from None
494 # Reset logging configuration
495 logging.getLogger().handlers = []
496 structlog.reset_defaults()
498 # Create a list of handlers
499 handlers = []
501 # Add console handler with Windows encoding support
503 # Use Windows-safe console handler for all platforms
504 console_handler = WindowsSafeConsoleHandler()
506 if clean_output and format == "console":
507 # Use clean formatter for console output
508 console_handler.setFormatter(CleanFormatter(use_custom_renderer=True))
509 else:
510 console_handler.setFormatter(logging.Formatter("%(message)s"))
512 console_handler.addFilter(ApplicationFilter()) # Only show our application logs
513 console_handler.addFilter(SQLiteFilter()) # Suppress verbose SQLite logs
515 if clean_output:
516 console_handler.addFilter(VerbosityFilter()) # Reduce verbosity
518 handlers.append(console_handler)
520 # Add file handler if file is configured
521 if file:
522 file_handler = CleanFileHandler(file)
523 file_handler.setFormatter(FileFormatter())
524 file_handler.addFilter(
525 SQLiteFilter()
526 ) # Suppress verbose SQLite logs in files too
527 # Don't apply verbosity filter to file logs - keep everything for debugging
528 handlers.append(file_handler)
530 # Configure standard logging
531 logging.basicConfig(
532 level=numeric_level,
533 format="%(message)s",
534 handlers=handlers,
535 )
537 # Apply SQLite filter to all existing loggers
538 root_logger = logging.getLogger()
539 for handler in root_logger.handlers:
540 handler.addFilter(SQLiteFilter())
542 # uvicorn.access has propagate=False in uvicorn's own logging config, so
543 # it bypasses the root logger's redaction above. Attach directly so
544 # webhook secrets/tokens in the request query string aren't logged in
545 # plaintext (see UvicornAccessRedactFilter docstring for details).
546 uvicorn_access_logger = logging.getLogger("uvicorn.access")
547 if not any(
548 isinstance(f, UvicornAccessRedactFilter)
549 for f in uvicorn_access_logger.filters
550 ):
551 uvicorn_access_logger.addFilter(UvicornAccessRedactFilter())
553 # Add filter to suppress Qdrant version check warnings
554 if suppress_qdrant_warnings:
555 qdrant_logger = logging.getLogger("qdrant_client")
556 qdrant_logger.addFilter(QdrantVersionFilter())
558 # Suppress verbose SQLAlchemy and SQLite logs
559 sqlalchemy_loggers = [
560 "sqlalchemy.engine",
561 "sqlalchemy.dialects",
562 "sqlalchemy.pool",
563 "aiosqlite",
564 "sqlite3",
565 ]
567 for logger_name in sqlalchemy_loggers:
568 logger = logging.getLogger(logger_name)
569 logger.addFilter(SQLiteFilter())
570 logger.setLevel(logging.WARNING) # Only show warnings and errors
572 # Suppress verbose third-party library debug logs
573 noisy_loggers = [
574 "chardet", # Character encoding detection
575 "chardet.charsetprober",
576 "chardet.latin1prober",
577 "chardet.mbcharsetprober",
578 "chardet.sbcharsetprober",
579 "chardet.utf8prober",
580 "pdfminer", # PDF parsing
581 "pdfminer.pdfparser",
582 "pdfminer.pdfdocument",
583 "pdfminer.pdfinterp",
584 "pdfminer.converter",
585 "pdfplumber", # PDF processing
586 "markitdown", # File conversion
587 "httpcore", # HTTP client debug logs
588 "httpx", # HTTP client debug logs
589 "gensim", # Topic modeling library
590 "gensim.models", # LDA model training logs
591 "gensim.models.ldamodel", # Specific LDA training logs
592 "gensim.corpora", # Dictionary and corpus logs
593 ]
595 for logger_name in noisy_loggers:
596 logger = logging.getLogger(logger_name)
597 logger.setLevel(logging.WARNING) # Only show warnings and errors
599 # Configure structlog processors based on format and clean_output
600 # Redaction processor to mask sensitive fields in event_dict
601 def _redact_sensitive(logger, method_name, event_dict): # type: ignore[no-redef]
602 sensitive_keys = {
603 "api_key",
604 "llm_api_key",
605 "authorization",
606 "Authorization",
607 "token",
608 "access_token",
609 "secret",
610 "password",
611 }
612 sensitive_key_pattern = re.compile(
613 r"(?i)(?:^|[_-])(token|secret|password|api[_-]?key|access[_-]?key|private[_-]?key|authorization)(?:$|[_-])"
614 )
616 def _is_sensitive_key(key: object) -> bool:
617 if not isinstance(key, str):
618 return False
619 return key in sensitive_keys or bool(sensitive_key_pattern.search(key))
621 def _mask(value: str) -> str:
622 try:
623 if not isinstance(value, str) or not value:
624 return "***REDACTED***"
625 # Preserve small hint while masking majority
626 if len(value) <= 8:
627 return "***REDACTED***"
628 return value[:2] + "***REDACTED***" + value[-2:]
629 except Exception:
630 return "***REDACTED***"
632 def _sanitize_string(value: str) -> str:
633 try:
634 return redact_sensitive_data(value, mask="***REDACTED***")
635 except Exception:
636 return "***REDACTED***"
638 def _deep_redact(obj):
639 try:
640 if isinstance(obj, dict):
641 red = {}
642 for k, v in obj.items():
643 if _is_sensitive_key(k):
644 red[k] = (
645 _mask(v) if isinstance(v, str) else "***REDACTED***"
646 )
647 else:
648 red[k] = _deep_redact(v)
649 return red
650 if isinstance(obj, list):
651 return [_deep_redact(i) for i in obj]
652 if isinstance(obj, str):
653 return _sanitize_string(obj)
654 return obj
655 except Exception:
656 return obj
658 return _deep_redact(event_dict)
660 if clean_output and format == "console":
661 # Minimal processors for clean output
662 processors = [
663 structlog.stdlib.filter_by_level,
664 structlog.processors.TimeStamper(fmt="%H:%M:%S"),
665 _redact_sensitive,
666 CustomConsoleRenderer(colors=True),
667 ]
668 else:
669 # Full processors for detailed output
670 processors = [
671 structlog.stdlib.filter_by_level,
672 structlog.stdlib.add_logger_name,
673 structlog.stdlib.add_log_level,
674 structlog.processors.TimeStamper(fmt="iso"),
675 structlog.processors.StackInfoRenderer(),
676 structlog.processors.UnicodeDecoder(),
677 structlog.processors.CallsiteParameterAdder(
678 [
679 structlog.processors.CallsiteParameter.FILENAME,
680 structlog.processors.CallsiteParameter.FUNC_NAME,
681 structlog.processors.CallsiteParameter.LINENO,
682 ]
683 ),
684 _redact_sensitive,
685 ]
687 if format == "json":
688 processors.append(structlog.processors.JSONRenderer())
689 else:
690 processors.append(structlog.dev.ConsoleRenderer(colors=True))
692 # Configure structlog
693 structlog.configure(
694 processors=processors,
695 wrapper_class=structlog.make_filtering_bound_logger(numeric_level),
696 logger_factory=structlog.stdlib.LoggerFactory(),
697 cache_logger_on_first_use=False, # Disable caching to ensure new configuration is used
698 )
700 cls._initialized = True
701 cls._current_config = (
702 level,
703 format,
704 file,
705 suppress_qdrant_warnings,
706 clean_output,
707 )
709 @classmethod
710 def get_logger(cls, name: str | None = None) -> structlog.BoundLogger:
711 """Get a logger instance.
713 Args:
714 name: Logger name. If None, will use the calling module's name.
716 Returns:
717 structlog.BoundLogger: Logger instance
718 """
719 if not cls._initialized:
720 # Initialize with default settings if not already initialized
721 cls.setup()
722 return structlog.get_logger(name)
725# Standardize on core logging configuration
726try:
727 from qdrant_loader_core.logging import (
728 LoggingConfig as _CoreLoggingConfig, # type: ignore
729 )
731 LoggingConfig = _CoreLoggingConfig # type: ignore[assignment]
732except Exception:
733 # Fallback to local implementation if core is unavailable
734 pass