repo_description: A production-ready FastAPI template for building AI agent applications with LangGraph integration. This template provides a robust foundation for building scalable, secure, and maintainable AI agent services.
repo_topics: agent agentic-ai docker fastapi fastapi-template langchain langchain-python langgraph langgraph-python llm memory
code: """Logging configuration and setup for the application.
  
  This module provides structured logging configuration using structlog,
  with environment-specific formatters and handlers. It supports both
  console-friendly development logging and JSON-formatted production logging.
  """
  
  import json
  import logging
  import sys
  from contextvars import ContextVar
  from datetime import datetime
  from pathlib import Path
  from typing import (
      Any,
      Dict,
      List,
      Optional,
  )
  
  import structlog
  from asgi_correlation_id import correlation_id
  
  from app.core.config import (
      Environment,
      settings,
  )
  
  # Ensure log directory exists
  settings.LOG_DIR.mkdir(parents=True, exist_ok=True)
  
  # Context variables for storing request-specific data
  _request_context: ContextVar[Dict[str, Any]] = ContextVar("request_context", default=None)
  
  
  def bind_context(**kwargs: Any) -> None:
      """Bind context variables to the current request.
  
      Args:
          **kwargs: Key-value pairs to bind to the logging context
      """
      current = _request_context.get() or {}
      _request_context.set({**current, **kwargs})
  
  
  def clear_context() -> None:
      """Clear all context variables for the current request."""
      _request_context.set(None)
  
  
  def get_context() -> Dict[str, Any]:
      """Get the current logging context.
  
      Returns:
          Dict[str, Any]: Current context dictionary
      """
      return _request_context.get() or {}
  
  
  def add_context_to_event_dict(logger: Any, method_name: str, event_dict: Dict[str, Any]) -> Dict[str, Any]:
      """Add context variables to the event dictionary.
  
      This processor adds any bound context variables to each log event.
  
      Args:
          logger: The logger instance
          method_name: The name of the logging method
          event_dict: The event dictionary to modify
  
      Returns:
          Dict[str, Any]: Modified event dictionary with context variables
      """
      context = get_context()
      if context:
          event_dict.update(context)
      return event_dict
  
  
  def add_request_id_to_event_dict(logger: Any, method_name: str, event_dict: Dict[str, Any]) -> Dict[str, Any]:
      """Add the current request_id (from asgi-correlation-id) to every log event.
  
      Args:
          logger: The logger instance
          method_name: The name of the logging method
          event_dict: The event dictionary to modify
  
      Returns:
          Dict[str, Any]: Modified event dictionary with request_id
      """
      request_id = correlation_id.get()
      if request_id:
          event_dict["request_id"] = request_id
      return event_dict
  
  
  def get_log_file_path() -> Path:
      """Get the current log file path based on date and environment.
  
      Returns:
          Path: The path to the log file
      """
      env_prefix = settings.ENVIRONMENT.value
      return settings.LOG_DIR / f"{env_prefix}-{datetime.now().strftime('%Y-%m-%d')}.jsonl"
  
  
  class JsonlFileHandler(logging.Handler):
      """Custom handler for writing JSONL logs to daily files."""
  
      def __init__(self, file_path: Path):
          """Initialize the JSONL file handler.
  
          Args:
              file_path: Path to the log file where entries will be written.
          """
          super().__init__()
          self.file_path = file_path
  
      def emit(self, record: logging.LogRecord) -> None:
          """Emit a record to the JSONL file."""
          try:
              log_entry = {
                  "timestamp": datetime.fromtimestamp(record.created).isoformat(),
                  "level": record.levelname,
                  "message": record.getMessage(),
                  "module": record.module,
                  "function": record.funcName,
                  "filename": record.pathname,
                  "line": record.lineno,
                  "environment": settings.ENVIRONMENT.value,
              }
              if hasattr(record, "extra"):
                  log_entry.update(record.extra)
  
              with open(self.file_path, "a", encoding="utf-8") as f:
                  f.write(json.dumps(log_entry) + "\n")
          except Exception:
              self.handleError(record)
  
      def close(self) -> None:
          """Close the handler."""
          super().close()
  
  
  def get_structlog_processors(include_file_info: bool = True) -> List[Any]:
      """Get the structlog processors based on configuration.
  
      Args:
          include_file_info: Whether to include file information in the logs
  
      Returns:
          List[Any]: List of structlog processors
      """
      # Set up processors that are common to both outputs
      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(),
          # Add context variables (user_id, session_id, etc.) to all log events
          add_context_to_event_dict,
          # Add request_id from asgi-correlation-id to all log events
          add_request_id_to_event_dict,
      ]
  
      # Add callsite parameters if file info is requested
      if include_file_info:
          processors.append(
              structlog.processors.CallsiteParameterAdder(
                  {
                      structlog.processors.CallsiteParameter.FILENAME,
                      structlog.processors.CallsiteParameter.FUNC_NAME,
                      structlog.processors.CallsiteParameter.LINENO,
                      structlog.processors.CallsiteParameter.MODULE,
                      structlog.processors.CallsiteParameter.PATHNAME,
                  }
              )
          )
  
      # Add environment info
      processors.append(lambda _, __, event_dict: {**event_dict, "environment": settings.ENVIRONMENT.value})
  
      return processors
  
  
  def setup_logging() -> None:
      """Configure structlog with different formatters based on environment.
  
      In development: pretty console output
      In staging/production: structured JSON logs
      """
      # Determine log level based on DEBUG setting
      log_level = logging.DEBUG if settings.DEBUG else logging.INFO
  
      # Create file handler for JSON logs
      file_handler = JsonlFileHandler(get_log_file_path())
      file_handler.setLevel(log_level)
  
      # Create console handler
      console_handler = logging.StreamHandler(sys.stdout)
      console_handler.setLevel(log_level)
  
      # Get shared processors
      shared_processors = get_structlog_processors(
          # Include detailed file info only in development and test
          include_file_info=settings.ENVIRONMENT in [Environment.DEVELOPMENT, Environment.TEST]
      )
  
      # Configure standard logging
      logging.basicConfig(
          format="%(message)s",
          level=log_level,
          handlers=[file_handler, console_handler],
      )
  
      # Configure structlog based on environment
      if settings.LOG_FORMAT == "console":
          # Development-friendly console logging
          structlog.configure(
              processors=[
                  *shared_processors,
                  # Use ConsoleRenderer for pretty output to the console
                  structlog.dev.ConsoleRenderer(),
              ],
              wrapper_class=structlog.stdlib.BoundLogger,
              logger_factory=structlog.stdlib.LoggerFactory(),
              cache_logger_on_first_use=True,
          )
      else:
          # Production JSON logging
          structlog.configure(
              processors=[
                  *shared_processors,
                  structlog.processors.JSONRenderer(),
              ],
              wrapper_class=structlog.stdlib.BoundLogger,
              logger_factory=structlog.stdlib.LoggerFactory(),
              cache_logger_on_first_use=True,
          )
  
  
  # Initialize logging
  setup_logging()
  
  # Create logger instance
  logger = structlog.get_logger()
  log_level_name = "DEBUG" if settings.DEBUG else "INFO"
  logger.info(
      "logging_initialized",
      environment=settings.ENVIRONMENT.value,
      log_level=log_level_name,
      log_format=settings.LOG_FORMAT,
      debug=settings.DEBUG,
  )
  