Coverage for sentimatrix / core / logger.py: 89%
212 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-01-28 09:30 +0000
1"""
2Sentimatrix Logging Module
4Provides structured logging with JSON output, context propagation,
5and multiple handlers (console, file).
7Example:
8 >>> from sentimatrix.core.logger import get_logger
9 >>> logger = get_logger(__name__)
10 >>> logger.info("Processing started", extra={"url": "https://example.com"})
11"""
13from __future__ import annotations
15import json
16import logging
17import sys
18from contextvars import ContextVar
19from datetime import datetime, timezone
20from logging.handlers import RotatingFileHandler
21from pathlib import Path
22from typing import Any, Dict, List, Optional, Union
24from rich.console import Console
25from rich.logging import RichHandler
26from rich.theme import Theme
28from sentimatrix.core.config import LogConfig, LogLevel
30# Context variable for request/correlation ID
31_request_context: ContextVar[Dict[str, Any]] = ContextVar("request_context", default={})
34class LogContext:
35 """
36 Context manager for adding contextual information to log messages.
38 Example:
39 >>> with LogContext(request_id="abc123", user_id="user456"):
40 ... logger.info("Processing request") # Includes request_id, user_id
41 """
43 def __init__(self, **kwargs: Any) -> None:
44 """Initialize context with key-value pairs."""
45 self._context = kwargs
46 self._token: Optional[Any] = None
48 def __enter__(self) -> "LogContext":
49 """Enter context and merge with existing context."""
50 current = _request_context.get().copy()
51 current.update(self._context)
52 self._token = _request_context.set(current)
53 return self
55 def __exit__(self, *args: Any) -> None:
56 """Exit context and restore previous context."""
57 if self._token is not None: 57 ↛ exitline 57 didn't return from function '__exit__' because the condition on line 57 was always true
58 _request_context.reset(self._token)
60 @staticmethod
61 def get() -> Dict[str, Any]:
62 """Get current context dictionary."""
63 return _request_context.get().copy()
65 @staticmethod
66 def set(**kwargs: Any) -> None:
67 """Set context values (replaces existing)."""
68 _request_context.set(kwargs)
70 @staticmethod
71 def update(**kwargs: Any) -> None:
72 """Update context values (merges with existing)."""
73 current = _request_context.get().copy()
74 current.update(kwargs)
75 _request_context.set(current)
77 @staticmethod
78 def clear() -> None:
79 """Clear all context."""
80 _request_context.set({})
83class JsonFormatter(logging.Formatter):
84 """
85 JSON log formatter for structured logging.
87 Produces JSON-formatted log messages with consistent structure:
88 {
89 "timestamp": "2024-01-15T12:00:00Z",
90 "level": "INFO",
91 "logger": "sentimatrix.core",
92 "message": "Processing started",
93 "caller": {"file": "main.py", "line": 42, "function": "process"},
94 "context": {"request_id": "abc123"},
95 "extra": {"url": "https://example.com"}
96 }
97 """
99 def __init__(
100 self,
101 include_timestamp: bool = True,
102 include_caller: bool = True,
103 timestamp_format: str = "%Y-%m-%dT%H:%M:%S.%fZ",
104 ) -> None:
105 """
106 Initialize JSON formatter.
108 Args:
109 include_timestamp: Include timestamp in output
110 include_caller: Include caller info (file, line, function)
111 timestamp_format: Format string for timestamps
112 """
113 super().__init__()
114 self.include_timestamp = include_timestamp
115 self.include_caller = include_caller
116 self.timestamp_format = timestamp_format
118 def format(self, record: logging.LogRecord) -> str:
119 """Format log record as JSON."""
120 # Build base log entry
121 log_entry: Dict[str, Any] = {
122 "level": record.levelname,
123 "logger": record.name,
124 "message": record.getMessage(),
125 }
127 # Add timestamp
128 if self.include_timestamp:
129 log_entry["timestamp"] = datetime.now(timezone.utc).strftime(self.timestamp_format)
131 # Add caller info
132 if self.include_caller:
133 log_entry["caller"] = {
134 "file": record.filename,
135 "line": record.lineno,
136 "function": record.funcName,
137 }
139 # Add context from ContextVar
140 context = LogContext.get()
141 if context:
142 log_entry["context"] = context
144 # Add extra fields (excluding standard LogRecord attributes)
145 standard_attrs = {
146 "name",
147 "msg",
148 "args",
149 "created",
150 "filename",
151 "funcName",
152 "levelname",
153 "levelno",
154 "lineno",
155 "module",
156 "msecs",
157 "pathname",
158 "process",
159 "processName",
160 "relativeCreated",
161 "stack_info",
162 "exc_info",
163 "exc_text",
164 "thread",
165 "threadName",
166 "taskName",
167 "message",
168 }
170 extra = {
171 key: value
172 for key, value in record.__dict__.items()
173 if key not in standard_attrs and not key.startswith("_")
174 }
176 if extra:
177 log_entry["extra"] = extra
179 # Add exception info if present
180 if record.exc_info:
181 log_entry["exception"] = self.formatException(record.exc_info)
183 return json.dumps(log_entry, default=str, ensure_ascii=False)
186class TextFormatter(logging.Formatter):
187 """
188 Human-readable text formatter with optional colorization.
190 Format: [TIMESTAMP] LEVEL - LOGGER - MESSAGE {extra}
191 """
193 COLORS = {
194 "DEBUG": "dim",
195 "INFO": "green",
196 "WARNING": "yellow",
197 "ERROR": "red",
198 "CRITICAL": "bold red",
199 }
201 def __init__(
202 self,
203 include_timestamp: bool = True,
204 include_caller: bool = False,
205 timestamp_format: str = "%Y-%m-%d %H:%M:%S",
206 ) -> None:
207 """
208 Initialize text formatter.
210 Args:
211 include_timestamp: Include timestamp in output
212 include_caller: Include caller info
213 timestamp_format: Format string for timestamps
214 """
215 super().__init__()
216 self.include_timestamp = include_timestamp
217 self.include_caller = include_caller
218 self.timestamp_format = timestamp_format
220 def format(self, record: logging.LogRecord) -> str:
221 """Format log record as text."""
222 parts: List[str] = []
224 # Timestamp
225 if self.include_timestamp:
226 timestamp = datetime.now().strftime(self.timestamp_format)
227 parts.append(f"[{timestamp}]")
229 # Level
230 parts.append(f"{record.levelname:8s}")
232 # Logger name (shortened)
233 logger_name = record.name
234 if logger_name.startswith("sentimatrix."):
235 logger_name = logger_name[12:] # Remove prefix
236 parts.append(f"{logger_name:20s}")
238 # Message
239 parts.append(record.getMessage())
241 # Caller info
242 if self.include_caller:
243 parts.append(f"({record.filename}:{record.lineno})")
245 # Context
246 context = LogContext.get()
247 if context: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 context_str = " ".join(f"{k}={v}" for k, v in context.items())
249 parts.append(f"[{context_str}]")
251 # Extra fields
252 standard_attrs = {
253 "name",
254 "msg",
255 "args",
256 "created",
257 "filename",
258 "funcName",
259 "levelname",
260 "levelno",
261 "lineno",
262 "module",
263 "msecs",
264 "pathname",
265 "process",
266 "processName",
267 "relativeCreated",
268 "stack_info",
269 "exc_info",
270 "exc_text",
271 "thread",
272 "threadName",
273 "taskName",
274 "message",
275 }
277 extra = {
278 key: value
279 for key, value in record.__dict__.items()
280 if key not in standard_attrs and not key.startswith("_")
281 }
283 if extra: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 extra_str = " ".join(f"{k}={v}" for k, v in extra.items())
285 parts.append(f"{{{extra_str}}}")
287 result = " ".join(parts)
289 # Add exception info
290 if record.exc_info: 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true
291 result += "\n" + self.formatException(record.exc_info)
293 return result
296class StructuredLogger:
297 """
298 Wrapper around Python logger with structured logging support.
300 Provides convenient methods for logging with extra context.
301 """
303 def __init__(self, logger: logging.Logger) -> None:
304 """Initialize with a Python logger instance."""
305 self._logger = logger
307 @property
308 def name(self) -> str:
309 """Get logger name."""
310 return self._logger.name
312 @property
313 def level(self) -> int:
314 """Get effective log level."""
315 return self._logger.getEffectiveLevel()
317 def _log(
318 self,
319 level: int,
320 message: str,
321 *args: Any,
322 exc_info: bool = False,
323 stack_info: bool = False,
324 extra: Optional[Dict[str, Any]] = None,
325 **kwargs: Any,
326 ) -> None:
327 """Internal log method with extra handling."""
328 # Merge kwargs into extra
329 if kwargs:
330 extra = extra or {}
331 extra.update(kwargs)
333 self._logger.log(
334 level,
335 message,
336 *args,
337 exc_info=exc_info,
338 stack_info=stack_info,
339 extra=extra,
340 )
342 def debug(
343 self,
344 message: str,
345 *args: Any,
346 extra: Optional[Dict[str, Any]] = None,
347 **kwargs: Any,
348 ) -> None:
349 """Log a debug message."""
350 self._log(logging.DEBUG, message, *args, extra=extra, **kwargs)
352 def info(
353 self,
354 message: str,
355 *args: Any,
356 extra: Optional[Dict[str, Any]] = None,
357 **kwargs: Any,
358 ) -> None:
359 """Log an info message."""
360 self._log(logging.INFO, message, *args, extra=extra, **kwargs)
362 def warning(
363 self,
364 message: str,
365 *args: Any,
366 extra: Optional[Dict[str, Any]] = None,
367 **kwargs: Any,
368 ) -> None:
369 """Log a warning message."""
370 self._log(logging.WARNING, message, *args, extra=extra, **kwargs)
372 def error(
373 self,
374 message: str,
375 *args: Any,
376 exc_info: bool = False,
377 extra: Optional[Dict[str, Any]] = None,
378 **kwargs: Any,
379 ) -> None:
380 """Log an error message."""
381 self._log(logging.ERROR, message, *args, exc_info=exc_info, extra=extra, **kwargs)
383 def critical(
384 self,
385 message: str,
386 *args: Any,
387 exc_info: bool = False,
388 extra: Optional[Dict[str, Any]] = None,
389 **kwargs: Any,
390 ) -> None:
391 """Log a critical message."""
392 self._log(logging.CRITICAL, message, *args, exc_info=exc_info, extra=extra, **kwargs)
394 def exception(
395 self,
396 message: str,
397 *args: Any,
398 extra: Optional[Dict[str, Any]] = None,
399 **kwargs: Any,
400 ) -> None:
401 """Log an exception with traceback."""
402 self._log(logging.ERROR, message, *args, exc_info=True, extra=extra, **kwargs)
404 def bind(self, **kwargs: Any) -> "StructuredLogger":
405 """
406 Create a new logger with bound context.
408 The bound context will be included in all log messages from the new logger.
410 Example:
411 >>> request_logger = logger.bind(request_id="abc123")
412 >>> request_logger.info("Processing") # Includes request_id
413 """
414 return BoundLogger(self._logger, kwargs)
417class BoundLogger(StructuredLogger):
418 """Logger with bound context that's included in every log message."""
420 def __init__(self, logger: logging.Logger, bindings: Dict[str, Any]) -> None:
421 """Initialize with base logger and bindings."""
422 super().__init__(logger)
423 self._bindings = bindings
425 def _log(
426 self,
427 level: int,
428 message: str,
429 *args: Any,
430 exc_info: bool = False,
431 stack_info: bool = False,
432 extra: Optional[Dict[str, Any]] = None,
433 **kwargs: Any,
434 ) -> None:
435 """Log with bound context merged into extra."""
436 merged_extra = self._bindings.copy()
437 if extra:
438 merged_extra.update(extra)
439 if kwargs: 439 ↛ 440line 439 didn't jump to line 440 because the condition on line 439 was never true
440 merged_extra.update(kwargs)
442 self._logger.log(
443 level,
444 message,
445 *args,
446 exc_info=exc_info,
447 stack_info=stack_info,
448 extra=merged_extra,
449 )
451 def bind(self, **kwargs: Any) -> "BoundLogger":
452 """Create a new bound logger with additional bindings."""
453 merged = self._bindings.copy()
454 merged.update(kwargs)
455 return BoundLogger(self._logger, merged)
458class LogManager:
459 """
460 Central log manager for configuring and creating loggers.
462 Manages logging configuration, handlers, and logger instances.
463 """
465 _instance: Optional["LogManager"] = None
466 _initialized: bool = False
468 def __new__(cls) -> "LogManager":
469 """Singleton pattern."""
470 if cls._instance is None:
471 cls._instance = super().__new__(cls)
472 return cls._instance
474 def __init__(self) -> None:
475 """Initialize log manager."""
476 if not LogManager._initialized:
477 self._loggers: Dict[str, StructuredLogger] = {}
478 self._config: Optional[LogConfig] = None
479 self._handlers: List[logging.Handler] = []
480 self._root_logger = logging.getLogger("sentimatrix")
481 LogManager._initialized = True
483 def configure(self, config: Optional[LogConfig] = None) -> None:
484 """
485 Configure logging based on LogConfig.
487 Args:
488 config: Logging configuration (uses defaults if None)
489 """
490 self._config = config or LogConfig()
492 # Clear existing handlers
493 for handler in self._handlers: 493 ↛ 494line 493 didn't jump to line 494 because the loop on line 493 never started
494 self._root_logger.removeHandler(handler)
495 self._handlers.clear()
497 # Set log level
498 level = getattr(logging, self._config.level.value)
499 self._root_logger.setLevel(level)
501 # Create formatter based on format type
502 if self._config.format == "json":
503 formatter = JsonFormatter(
504 include_timestamp=self._config.include_timestamp,
505 include_caller=self._config.include_caller,
506 )
507 else:
508 formatter = TextFormatter(
509 include_timestamp=self._config.include_timestamp,
510 include_caller=self._config.include_caller,
511 )
513 # Console handler
514 if self._config.console_output: 514 ↛ 515line 514 didn't jump to line 515 because the condition on line 514 was never true
515 if self._config.colorize and self._config.format == "text":
516 # Use Rich for colorized console output
517 console = Console(
518 theme=Theme(
519 {
520 "logging.level.debug": "dim",
521 "logging.level.info": "green",
522 "logging.level.warning": "yellow",
523 "logging.level.error": "red",
524 "logging.level.critical": "bold red",
525 }
526 )
527 )
528 console_handler = RichHandler(
529 console=console,
530 show_time=self._config.include_timestamp,
531 show_path=self._config.include_caller,
532 rich_tracebacks=True,
533 )
534 else:
535 console_handler = logging.StreamHandler(sys.stdout)
536 console_handler.setFormatter(formatter)
538 console_handler.setLevel(level)
539 self._root_logger.addHandler(console_handler)
540 self._handlers.append(console_handler)
542 # File handler
543 if self._config.file_path:
544 file_path = Path(self._config.file_path)
545 file_path.parent.mkdir(parents=True, exist_ok=True)
547 file_handler = RotatingFileHandler(
548 file_path,
549 maxBytes=self._config.max_file_size_mb * 1024 * 1024,
550 backupCount=self._config.backup_count,
551 encoding="utf-8",
552 )
553 file_handler.setFormatter(formatter)
554 file_handler.setLevel(level)
555 self._root_logger.addHandler(file_handler)
556 self._handlers.append(file_handler)
558 # Prevent propagation to root logger
559 self._root_logger.propagate = False
561 def get_logger(self, name: str) -> StructuredLogger:
562 """
563 Get or create a structured logger.
565 Args:
566 name: Logger name (typically __name__)
568 Returns:
569 StructuredLogger instance
570 """
571 if name not in self._loggers:
572 # Ensure logger is under sentimatrix namespace
573 if not name.startswith("sentimatrix"): 573 ↛ 576line 573 didn't jump to line 576 because the condition on line 573 was always true
574 full_name = f"sentimatrix.{name}"
575 else:
576 full_name = name
578 logger = logging.getLogger(full_name)
579 self._loggers[name] = StructuredLogger(logger)
581 return self._loggers[name]
583 def set_level(self, level: Union[str, LogLevel]) -> None:
584 """Set log level for all loggers."""
585 if isinstance(level, LogLevel):
586 level = level.value
587 log_level = getattr(logging, level.upper())
588 self._root_logger.setLevel(log_level)
589 for handler in self._handlers: 589 ↛ 590line 589 didn't jump to line 590 because the loop on line 589 never started
590 handler.setLevel(log_level)
592 def shutdown(self) -> None:
593 """Shutdown logging and cleanup handlers."""
594 for handler in self._handlers:
595 handler.close()
596 self._root_logger.removeHandler(handler)
597 self._handlers.clear()
598 self._loggers.clear()
601# Module-level convenience functions
604def get_logger(name: str = "sentimatrix") -> StructuredLogger:
605 """
606 Get a structured logger instance.
608 Args:
609 name: Logger name (typically __name__)
611 Returns:
612 StructuredLogger instance
614 Example:
615 >>> logger = get_logger(__name__)
616 >>> logger.info("Processing started", url="https://example.com")
617 """
618 manager = LogManager()
619 if not manager._config: 619 ↛ 620line 619 didn't jump to line 620 because the condition on line 619 was never true
620 manager.configure()
621 return manager.get_logger(name)
624def configure_logging(config: Optional[LogConfig] = None) -> None:
625 """
626 Configure logging globally.
628 Args:
629 config: Logging configuration (uses defaults if None)
631 Example:
632 >>> from sentimatrix.core.config import LogConfig, LogLevel
633 >>> configure_logging(LogConfig(level=LogLevel.DEBUG, format="text"))
634 """
635 manager = LogManager()
636 manager.configure(config)
639def set_log_level(level: Union[str, LogLevel]) -> None:
640 """
641 Set global log level.
643 Args:
644 level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
646 Example:
647 >>> set_log_level("DEBUG")
648 >>> set_log_level(LogLevel.WARNING)
649 """
650 manager = LogManager()
651 manager.set_level(level)