""" Structured logging configuration using structlog. Deeply integrates with FastAPI/uvicorn's built-in logging to provide seamless structured logs across the entire application stack. """ import logging import logging.config import sys from contextlib import asynccontextmanager from datetime import datetime, timezone from typing import Any, AsyncIterator import structlog from structlog.types import EventDict, Processor from .config import config def add_timestamp(logger: Any, method_name: str, event_dict: EventDict) -> EventDict: """Add ISO 8601 timestamp to log entries.""" event_dict["timestamp"] = datetime.now(timezone.utc).isoformat() return event_dict def add_log_level(logger: Any, method_name: str, event_dict: EventDict) -> EventDict: """Add log level to event dict.""" event_dict["level"] = method_name.upper() return event_dict def extract_from_record(logger: Any, method_name: str, event_dict: EventDict) -> EventDict: """ Extract extra fields from logging.LogRecord for standard library integration. This allows standard Python logging calls to include structured data: logger.info("request received", extra={"user_id": "123", "path": "/api"}) """ record = event_dict.get("_record") if record is not None: # Extract custom fields from record for key, value in record.__dict__.items(): if key not in { "name", "msg", "args", "created", "filename", "funcName", "levelname", "levelno", "lineno", "module", "msecs", "message", "pathname", "process", "processName", "relativeCreated", "thread", "threadName", "exc_info", "exc_text", "stack_info", "taskName" }: event_dict[key] = value return event_dict def configure_logging() -> None: """ Configure structured logging with deep FastAPI/uvicorn integration. - Replaces all Python logging with structlog - FastAPI, uvicorn, and app logs all use same format - JSON format for production, pretty console for development - Preserves log levels and exception handling """ # Determine processors based on log format shared_processors: list[Processor] = [ structlog.contextvars.merge_contextvars, structlog.stdlib.add_logger_name, add_log_level, add_timestamp, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.StackInfoRenderer(), extract_from_record, ] if config.log_format == "json": # JSON format for production structlog.configure( processors=[ structlog.stdlib.filter_by_level, *shared_processors, structlog.stdlib.ProcessorFormatter.wrap_for_formatter, ], logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) formatter = structlog.stdlib.ProcessorFormatter( processors=[ structlog.stdlib.ProcessorFormatter.remove_processors_meta, structlog.processors.format_exc_info, structlog.processors.JSONRenderer(), ], foreign_pre_chain=shared_processors, ) else: # Console format for development structlog.configure( processors=[ structlog.stdlib.filter_by_level, *shared_processors, structlog.stdlib.ProcessorFormatter.wrap_for_formatter, ], logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) formatter = structlog.stdlib.ProcessorFormatter( processors=[ structlog.stdlib.ProcessorFormatter.remove_processors_meta, structlog.dev.ConsoleRenderer(colors=True), ], foreign_pre_chain=shared_processors, ) # Configure Python's logging to use structlog handler = logging.StreamHandler(sys.stdout) handler.setFormatter(formatter) # Set up root logger root_logger = logging.getLogger() root_logger.handlers.clear() root_logger.addHandler(handler) root_logger.setLevel(logging.getLevelName(config.effective_log_level)) # Configure specific loggers for logger_name in [ "uvicorn", "uvicorn.access", "uvicorn.error", "fastapi", "tatlock", ]: logger = logging.getLogger(logger_name) logger.handlers.clear() logger.propagate = True logger.setLevel(logging.getLevelName(config.effective_log_level)) def get_logger(name: str) -> structlog.stdlib.BoundLogger: """ Get a structured logger instance. Works seamlessly with both structlog and standard logging calls: - logger.info("message", key="value") - structlog style - logger.info("message", extra={"key": "value"}) - standard logging style Args: name: Logger name (typically __name__) Returns: Configured structlog BoundLogger Example: >>> logger = get_logger(__name__) >>> logger.info("user_request", user_id="123", action="search") >>> logger.info("standard log", extra={"request_id": "abc"}) """ return structlog.get_logger(name) @asynccontextmanager async def log_operation( operation: str, initial_context: dict[str, Any] | None = None, logger_name: str = "tatlock.operations" ) -> AsyncIterator[dict[str, Any]]: """ Context manager for automatic operation timing and logging. Args: operation: Operation name (e.g., "steward_analysis", "tool_call") initial_context: Initial metadata to log logger_name: Logger name for this operation Yields: Context dict that can be updated during operation Example: >>> async with log_operation("steward_analysis", {"user_id": "123"}) as ctx: ... # Do work ... ctx["recommendation_count"] = 3 ... # Automatically logs duration and context on exit """ logger = get_logger(logger_name) context = initial_context or {} context["operation"] = operation start_time = datetime.now(timezone.utc) logger.info("operation_started", **context) try: yield context # Success case duration = (datetime.now(timezone.utc) - start_time).total_seconds() context["duration_seconds"] = duration context["success"] = True logger.info("operation_completed", **context) except Exception as e: # Error case duration = (datetime.now(timezone.utc) - start_time).total_seconds() context["duration_seconds"] = duration context["success"] = False context["error"] = str(e) context["error_type"] = type(e).__name__ logger.error("operation_failed", **context, exc_info=True) raise def get_uvicorn_log_config() -> dict[str, Any]: """ Get uvicorn logging configuration that integrates with structlog. Use this when starting uvicorn: uvicorn.run(app, log_config=get_uvicorn_log_config()) Returns: Uvicorn-compatible logging configuration dict """ return { "version": 1, "disable_existing_loggers": False, "formatters": { "default": { "()": structlog.stdlib.ProcessorFormatter, "processors": [ structlog.stdlib.ProcessorFormatter.remove_processors_meta, structlog.processors.JSONRenderer() if config.log_format == "json" else structlog.dev.ConsoleRenderer(colors=True), ], }, }, "handlers": { "default": { "formatter": "default", "class": "logging.StreamHandler", "stream": "ext://sys.stdout", }, }, "loggers": { "uvicorn": {"handlers": ["default"], "level": config.effective_log_level}, "uvicorn.error": {"handlers": ["default"], "level": config.effective_log_level}, "uvicorn.access": {"handlers": ["default"], "level": config.effective_log_level}, }, } # Initialize logging on module import configure_logging()