| """Structured logging middleware with structlog.""" |
|
|
| import logging |
| import structlog |
| from functools import wraps |
| from typing import Callable, Any |
| import time |
|
|
|
|
| def configure_logging(): |
| """Configure structured logging.""" |
| logging.basicConfig(level=logging.INFO) |
| structlog.configure( |
| processors=[ |
| structlog.stdlib.filter_by_level, |
| structlog.stdlib.add_logger_name, |
| structlog.stdlib.add_log_level, |
| structlog.stdlib.PositionalArgumentsFormatter(), |
| structlog.processors.TimeStamper(fmt="iso"), |
| structlog.processors.StackInfoRenderer(), |
| structlog.processors.format_exc_info, |
| structlog.processors.UnicodeDecoder(), |
| structlog.processors.JSONRenderer() |
| ], |
| context_class=dict, |
| logger_factory=structlog.stdlib.LoggerFactory(), |
| cache_logger_on_first_use=True, |
| ) |
|
|
|
|
| def get_logger(name: str) -> structlog.stdlib.BoundLogger: |
| """Get a configured logger.""" |
| return structlog.get_logger(name) |
|
|
|
|
| def log_execution(logger: structlog.stdlib.BoundLogger): |
| """Decorator to log function execution.""" |
| def decorator(func: Callable) -> Callable: |
| @wraps(func) |
| async def async_wrapper(*args, **kwargs) -> Any: |
| start_time = time.time() |
| logger.info(f"Starting {func.__name__}") |
| try: |
| result = await func(*args, **kwargs) |
| duration = time.time() - start_time |
| logger.info(f"Completed {func.__name__}", duration=duration) |
| return result |
| except Exception as e: |
| duration = time.time() - start_time |
| logger.error(f"Error in {func.__name__}", error=str(e), duration=duration) |
| raise |
|
|
| @wraps(func) |
| def sync_wrapper(*args, **kwargs) -> Any: |
| start_time = time.time() |
| logger.info(f"Starting {func.__name__}") |
| try: |
| result = func(*args, **kwargs) |
| duration = time.time() - start_time |
| logger.info(f"Completed {func.__name__}", duration=duration) |
| return result |
| except Exception as e: |
| duration = time.time() - start_time |
| logger.error(f"Error in {func.__name__}", error=str(e), duration=duration) |
| raise |
|
|
| return async_wrapper if hasattr(func, '__call__') and hasattr(func, '__code__') and func.__code__.co_flags & 0x80 else sync_wrapper |
| return decorator |
|
|
