"""Sets up global logging configuration for the application.""" import logging import os logger = logging.getLogger(__name__) def configure_logging(): """ Called at startup of a FastAPI application instance to setup logging. Depending on the environment, it will log to stdout or to Loki. """ is_debug = os.getenv('DEBUG', "false") == "true" is_kubernetes = os.getenv('KUBERNETES_SERVICE_HOST') is not None if is_kubernetes: # in that case we want to log to stdout and also to loki from loki_logger_handler.loki_logger_handler import LokiLoggerHandler loki_url = os.getenv('LOKI_URL') if loki_url is None: raise ValueError("LOKI_URL environment variable is not set") loki_handler = LokiLoggerHandler( url = loki_url, labels = {'app': 'anyway', 'environment': 'staging' if is_debug else 'production'} ) logger.info(f"Logging to Loki at {loki_url} with {loki_handler.labels} and {is_debug=}") logging_handlers = [loki_handler, logging.StreamHandler()] logging_level = logging.DEBUG if is_debug else logging.INFO # silence the chatty logs loki generates itself logging.getLogger('urllib3.connectionpool').setLevel(logging.WARNING) # no need for time since it's added by loki or can be shown in kube logs logging_format = '%(name)s - %(levelname)s - %(message)s' else: # if we are in a debug (local) session, set verbose and rich logging from rich.logging import RichHandler logging_handlers = [RichHandler()] logging_level = logging.DEBUG if is_debug else logging.INFO logging_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' logging.basicConfig( level = logging_level, format = logging_format, handlers = logging_handlers ) # also overwrite the uvicorn loggers logging.getLogger('uvicorn').handlers = logging_handlers logging.getLogger('uvicorn.access').handlers = logging_handlers logging.getLogger('uvicorn.error').handlers = logging_handlers