logger.py•2.94 kB
"""
Structured logging configuration using structlog.
"""
import logging
import sys
from pathlib import Path
from typing import Any, Dict, Optional
import structlog
from .settings import LogFormat, LogLevel
def configure_logging(
level: LogLevel = LogLevel.INFO,
format: LogFormat = LogFormat.JSON,
log_file: Optional[Path] = None,
) -> structlog.BoundLogger:
"""
Configure structured logging.
Args:
level: Logging level
format: Log format (json or text)
log_file: Optional log file path
Returns:
Configured logger instance
"""
# Convert string level to logging level
log_level = getattr(logging, level.value)
# Configure processors based on format
if format == LogFormat.JSON:
processors = [
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer(),
]
else:
processors = [
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.dev.ConsoleRenderer(),
]
# Configure structlog
structlog.configure(
processors=processors,
wrapper_class=structlog.make_filtering_bound_logger(log_level),
context_class=dict,
logger_factory=structlog.PrintLoggerFactory(),
cache_logger_on_first_use=True,
)
# Configure standard logging to redirect to structlog
logging.basicConfig(
format="%(message)s",
stream=sys.stdout,
level=log_level,
)
# Add file handler if log_file is specified
if log_file:
log_file.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(log_level)
logging.getLogger().addHandler(file_handler)
return structlog.get_logger()
def get_logger(name: Optional[str] = None, **initial_values: Any) -> structlog.BoundLogger:
"""
Get a logger instance with optional name and initial context.
Args:
name: Logger name (usually module name)
**initial_values: Initial context values to bind to the logger
Returns:
Bound logger instance
"""
logger = structlog.get_logger(name)
if initial_values:
logger = logger.bind(**initial_values)
return logger
def log_context(**kwargs: Any) -> Dict[str, Any]:
"""
Create a context dictionary for logging.
Args:
**kwargs: Context key-value pairs
Returns:
Context dictionary
"""
return kwargs