Coverage for src/taipanstack/utils/logging.py: 100%
180 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-10-13 20:52 +0000
1"""
2Structured logging with context.
4Provides a configured logger with support for structured output,
5context propagation, and proper formatting.
6"""
8import logging
9import logging as _std_logging
10import re
11import sys
12from collections.abc import Iterator, MutableMapping
13from contextlib import AbstractContextManager, contextmanager
14from datetime import UTC, datetime
15from functools import lru_cache
16from typing import Literal, TypeVar, cast
18from taipanstack.utils.context import get_correlation_id
20try:
21 import structlog
23 HAS_STRUCTLOG = True
24except ImportError:
25 HAS_STRUCTLOG = False
28# Default log format
29DEFAULT_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
30JSON_FORMAT = (
31 '{"timestamp": "%(asctime)s", "level": "%(levelname)s", '
32 '"logger": "%(name)s", "message": "%(message)s"}'
33)
35SENSITIVE_KEY_PATTERNS: tuple[str, ...] = (
36 "password",
37 "secret",
38 "token",
39 "authorization",
40 "api_key",
41)
43# Pre-compile the regex for O(1) checks per key instead of O(N*M)
44_SENSITIVE_KEY_REGEX = (
45 re.compile("|".join(map(re.escape, SENSITIVE_KEY_PATTERNS)), re.IGNORECASE)
46 if SENSITIVE_KEY_PATTERNS
47 else None
48)
50REDACTED_VALUE = "***REDACTED***"
53@lru_cache(maxsize=1024) # type: ignore[misc]
54def _is_sensitive(key: object, regex: re.Pattern[str] | None) -> bool:
55 """Check if a key is sensitive using cached regex matching.
57 Args:
58 key: The key to check.
59 regex: The regex pattern to use for matching.
61 Returns:
62 True if the key is a string and sensitive, False otherwise.
64 """
65 if regex is None:
66 return False
67 if not isinstance(key, str):
68 return False
69 return bool(regex.search(key))
72KT = TypeVar("KT")
73VT = TypeVar("VT")
76def _redact_mapping(
77 obj: MutableMapping[KT, VT],
78 seen: set[int],
79) -> dict[KT, object]:
80 """Redact a mapping object."""
81 seen.add(id(obj))
82 return {
83 k: (
84 REDACTED_VALUE
85 if _is_sensitive(k, _SENSITIVE_KEY_REGEX)
86 else _redact(v, seen)
87 )
88 for k, v in obj.items()
89 }
92def _redact_list(obj: list[VT], seen: set[int]) -> list[object]:
93 """Redact a list object."""
94 seen.add(id(obj))
95 return [_redact(item, seen) for item in obj]
98def _redact_tuple(obj: tuple[VT, ...], seen: set[int]) -> tuple[object, ...]:
99 """Redact a tuple object."""
100 seen.add(id(obj))
101 return tuple(_redact(item, seen) for item in obj)
104def _is_unhashable(item: object) -> bool:
105 try:
106 hash(item)
107 return False
108 except TypeError:
109 return True
112def _redact_set(obj: set[VT], seen: set[int]) -> set[object] | list[object]:
113 """Redact a set object."""
114 seen.add(id(obj))
115 redacted_items = [_redact(item, seen) for item in obj]
116 if any(_is_unhashable(item) for item in redacted_items):
117 return redacted_items
118 return set(redacted_items)
121def _redact_iterable(obj: object, seen: set[int]) -> object:
122 """Dispatch redaction based on iterable type."""
123 if isinstance(obj, list):
124 return _redact_list(obj, seen)
125 if isinstance(obj, tuple):
126 return _redact_tuple(obj, seen)
127 if isinstance(obj, set):
128 return _redact_set(obj, seen)
129 return obj
132def _redact_collection(obj: object, seen: set[int]) -> object:
133 """Dispatch redaction based on collection type."""
134 if isinstance(obj, (dict, MutableMapping)):
135 return _redact_mapping(cast(MutableMapping[object, object], obj), seen)
137 return _redact_iterable(obj, seen)
140def _redact(obj: object, seen: set[int] | None = None) -> object:
141 """Redact sensitive data recursively with circular reference protection.
143 Args:
144 obj: The object to redact.
145 seen: Set of IDs of already processed containers.
147 Returns:
148 A new object with sensitive data redacted.
150 """
151 if _SENSITIVE_KEY_REGEX is None:
152 return obj
154 if seen is None:
155 seen = set()
157 # Detect circular references
158 if id(obj) in seen:
159 return REDACTED_VALUE
161 return _redact_collection(obj, seen)
164def _redact_dict(d: MutableMapping[str, object]) -> None:
165 """Redact sensitive keys in a dictionary in-place (top-level only).
167 Note: Nested structures are replaced with redacted copies to prevent
168 shared state mutation.
170 Args:
171 d: The dictionary to redact.
173 """
174 if _SENSITIVE_KEY_REGEX is None:
175 return
177 for key, value in list(d.items()):
178 if _is_sensitive(key, _SENSITIVE_KEY_REGEX):
179 d[key] = REDACTED_VALUE
180 else:
181 d[key] = _redact(value)
184def mask_sensitive_data_processor(
185 _logger: object,
186 _method: str,
187 event_dict: MutableMapping[str, object],
188) -> MutableMapping[str, object]:
189 """Mask sensitive data in structlog event dictionaries.
191 Intercept the *event_dict* produced by structlog and replace the
192 value of any key whose name contains a sensitive substring with
193 ``"***REDACTED***"``. Matching is case-insensitive.
195 Args:
196 _logger: The wrapped logger object (unused, required by structlog).
197 _method: The name of the log method called (unused, required by structlog).
198 event_dict: The structured event dictionary.
200 Returns:
201 The event dictionary with sensitive values masked.
203 """
204 _redact_dict(event_dict)
205 return event_dict
208def correlation_id_processor(
209 _logger: object,
210 _method: str,
211 event_dict: MutableMapping[str, object],
212) -> MutableMapping[str, object]:
213 """Structlog processor to inject correlation ID into events.
215 Args:
216 _logger: The wrapped logger object.
217 _method: The name of the log method called.
218 event_dict: The structured event dictionary.
220 Returns:
221 The event dictionary with correlation_id injected if sets.
223 """
224 correlation_id = get_correlation_id()
225 if correlation_id is not None:
226 event_dict["correlation_id"] = correlation_id
227 return event_dict
230class StackLogger:
231 """Enhanced logger with context support.
233 Provides a wrapper around standard logging with additional features
234 like context propagation and structured output support.
236 Attributes:
237 name: Logger name.
238 level: Current log level.
240 """
242 def __init__(
243 self,
244 name: str = "taipanstack",
245 level: str = "INFO",
246 *,
247 use_structured: bool = False,
248 ) -> None:
249 """Initialize the logger.
251 Args:
252 name: Logger name.
253 level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL).
254 use_structured: Use structured logging if structlog is available.
256 """
257 self.name = name
258 self.level = level
259 self._context: dict[str, object] = {}
261 if use_structured and HAS_STRUCTLOG:
262 self._logger = structlog.get_logger(name) # type: ignore[misc]
263 self._structured = True
264 else:
265 self._logger = logging.getLogger(name)
266 self._logger.setLevel(getattr(logging, level.upper())) # type: ignore[misc]
267 self._structured = False
269 def bind(self, **context: object) -> "StackLogger":
270 """Add context to logger.
272 Args:
273 **context: Key-value pairs to add to context.
275 Returns:
276 Self for chaining.
278 """
279 self._context.update(context)
280 if self._structured and HAS_STRUCTLOG:
281 self._logger = self._logger.bind(**context) # type: ignore[misc]
282 return self
284 def unbind(self, *keys: str) -> "StackLogger":
285 """Remove context keys.
287 Args:
288 *keys: Keys to remove from context.
290 Returns:
291 Self for chaining.
293 """
294 for key in keys:
295 self._context.pop(key, None)
296 if self._structured and HAS_STRUCTLOG:
297 self._logger = self._logger.unbind(*keys) # type: ignore[misc]
298 return self
300 def _format_message(self, message: str, **kwargs: object) -> str:
301 """Format message with context.
303 Args:
304 message: The log message.
305 **kwargs: Additional context for this message.
307 Returns:
308 Formatted message string.
310 """
311 if not kwargs and not self._context:
312 return message
314 context = {**self._context, **kwargs}
315 _redact_dict(context)
317 context_str = " ".join(f"{k}={v}" for k, v in context.items())
318 return f"{message} | {context_str}"
320 def debug(self, message: str, **kwargs: object) -> None:
321 """Log a debug message.
323 Args:
324 message: The message to log.
325 **kwargs: Additional context.
327 """
328 if self._structured:
329 self._logger.debug(message, **kwargs) # type: ignore[misc]
330 else:
331 self._logger.debug(self._format_message(message, **kwargs)) # type: ignore[misc]
333 def info(self, message: str, **kwargs: object) -> None:
334 """Log an info message.
336 Args:
337 message: The message to log.
338 **kwargs: Additional context.
340 """
341 if self._structured:
342 self._logger.info(message, **kwargs) # type: ignore[misc]
343 else:
344 self._logger.info(self._format_message(message, **kwargs)) # type: ignore[misc]
346 def warning(self, message: str, **kwargs: object) -> None:
347 """Log a warning message.
349 Args:
350 message: The message to log.
351 **kwargs: Additional context.
353 """
354 if self._structured:
355 self._logger.warning(message, **kwargs) # type: ignore[misc]
356 else:
357 self._logger.warning(self._format_message(message, **kwargs)) # type: ignore[misc]
359 def error(self, message: str, **kwargs: object) -> None:
360 """Log an error message.
362 Args:
363 message: The message to log.
364 **kwargs: Additional context.
366 """
367 if self._structured:
368 self._logger.error(message, **kwargs) # type: ignore[misc]
369 else:
370 self._logger.error(self._format_message(message, **kwargs)) # type: ignore[misc]
372 def critical(self, message: str, **kwargs: object) -> None:
373 """Log a critical message.
375 Args:
376 message: The message to log.
377 **kwargs: Additional context.
379 """
380 if self._structured:
381 self._logger.critical(message, **kwargs) # type: ignore[misc]
382 else:
383 self._logger.critical(self._format_message(message, **kwargs)) # type: ignore[misc]
385 def exception(self, message: str, **kwargs: object) -> None:
386 """Log an exception with traceback.
388 Args:
389 message: The message to log.
390 **kwargs: Additional context.
392 """
393 if self._structured:
394 self._logger.exception(message, **kwargs) # type: ignore[misc]
395 else:
396 self._logger.exception(self._format_message(message, **kwargs)) # type: ignore[misc]
399def _configure_structlog(*, level: int | str | None = None) -> None:
400 """Configure structlog with default processors and settings."""
401 if level is not None:
402 if isinstance(level, str):
403 level = getattr(_std_logging, level.upper(), _std_logging.INFO)
404 _std_logging.basicConfig(level=level) # type: ignore[misc]
406 structlog.configure(
407 processors=[
408 structlog.stdlib.filter_by_level,
409 structlog.stdlib.add_logger_name,
410 structlog.stdlib.add_log_level, # type: ignore[misc]
411 structlog.processors.TimeStamper(fmt="iso"),
412 correlation_id_processor,
413 mask_sensitive_data_processor,
414 structlog.processors.StackInfoRenderer(),
415 structlog.processors.format_exc_info,
416 structlog.processors.UnicodeDecoder(),
417 structlog.processors.JSONRenderer(),
418 ],
419 wrapper_class=structlog.stdlib.BoundLogger,
420 context_class=dict,
421 logger_factory=structlog.stdlib.LoggerFactory(),
422 )
425def _get_log_format(format_type: str) -> str:
426 """Get the appropriate log format string."""
427 if format_type == "simple":
428 return "%(levelname)s: %(message)s"
429 elif format_type == "json":
430 return JSON_FORMAT
431 else:
432 return DEFAULT_FORMAT
435def setup_logging(
436 level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
437 *,
438 format_type: Literal["simple", "detailed", "json"] = "detailed",
439 log_file: str | None = None,
440 use_structured: bool = False,
441) -> None:
442 """Configure the root logger.
444 Args:
445 level: Log level to set.
446 format_type: Output format type.
447 log_file: Optional file to log to.
448 use_structured: Use structlog if available.
450 """
451 # Configure structlog if available and requested
452 if use_structured and HAS_STRUCTLOG:
453 _configure_structlog()
454 return
456 log_format = _get_log_format(format_type)
458 handlers: list[logging.Handler] = [logging.StreamHandler(sys.stdout)]
460 if log_file:
461 handlers.append(logging.FileHandler(log_file, encoding="utf-8"))
463 logging.basicConfig(
464 level=getattr(logging, level.upper()), # type: ignore[misc]
465 format=log_format,
466 handlers=handlers,
467 force=True,
468 )
471def get_logger(
472 name: str = "taipanstack",
473 *,
474 level: str = "INFO",
475 use_structured: bool = False,
476) -> StackLogger:
477 """Get a configured logger instance.
479 Args:
480 name: Logger name.
481 level: Log level.
482 use_structured: Use structlog if available.
484 Returns:
485 Configured StackLogger instance.
487 Example:
488 >>> logger = get_logger("my_module")
489 >>> logger.bind(request_id="123").info("Processing request")
491 """
492 return StackLogger(name, level, use_structured=use_structured)
495def log_operation(
496 operation: str,
497 *,
498 logger: StackLogger | None = None,
499 level: str = "INFO",
500 expected_exceptions: tuple[type[Exception], ...] | type[Exception] = Exception,
501) -> AbstractContextManager[StackLogger]:
502 """Context manager for logging operations.
504 Args:
505 operation: Name of the operation.
506 logger: Logger to use (creates one if not provided).
507 level: Log level for messages.
508 expected_exceptions: Exceptions to catch and log as failures.
510 Yields:
511 The logger instance.
513 Example:
514 >>> with log_operation("setup") as logger:
515 ... logger.info("Setting up environment")
517 """
519 @contextmanager
520 def _log_context() -> Iterator[StackLogger]:
521 nonlocal logger
522 if logger is None:
523 logger = get_logger()
525 start_time = datetime.now(UTC)
526 logger.bind(operation=operation)
528 log_method = getattr(logger, level.lower()) # type: ignore[misc]
529 log_method(f"Starting: {operation}")
531 try:
532 yield logger
533 duration = (datetime.now(UTC) - start_time).total_seconds()
534 log_method(f"Completed: {operation}", duration_seconds=duration)
535 except expected_exceptions as e:
536 duration = (datetime.now(UTC) - start_time).total_seconds()
537 logger.exception(
538 f"Failed: {operation}",
539 duration_seconds=duration,
540 error=str(e),
541 )
542 raise
543 finally:
544 logger.unbind("operation")
546 return _log_context()