import logging import sys import structlog from structlog.types import EventDict, Processor # https://github.com/hynek/structlog/issues/35#issuecomment-591321744 def rename_event_key(_, __, event_dict: EventDict) -> EventDict: """ Log entries keep the text message in the `event` field, but Datadog uses the `message` field. This processor moves the value from one field to the other. See https://github.com/hynek/structlog/issues/35#issuecomment-591321744 """ event_dict["message"] = event_dict.pop("event") return event_dict def drop_color_message_key(_, __, event_dict: EventDict) -> EventDict: """ Uvicorn logs the message a second time in the extra `color_message`, but we don't need it. This processor drops the key from the event dict if it exists. """ event_dict.pop("color_message", None) return event_dict def setup_logging(json_logs: bool = False, log_level: str = "INFO"): """Enhance the configuration of structlog. Needed for correlation id injection with fastapi middleware in samgis-web. After the use of logging_middleware() in samgis_web.web.middlewares, add also the CorrelationIdMiddleware from 'asgi_correlation_id' package. (See 'tests/web/test_middlewares.py' in samgis_web). To change an input parameter like the log level, re-run the function changing the parameter (no need to re-instantiate the logger instance: it's a hot change) Args: json_logs: set logs in json format log_level: log level string Returns: """ timestamper = structlog.processors.TimeStamper(fmt="iso") shared_processors: list[Processor] = [ structlog.contextvars.merge_contextvars, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.stdlib.ExtraAdder(), drop_color_message_key, timestamper, structlog.processors.StackInfoRenderer(), # adapted from https://www.structlog.org/en/stable/standard-library.html # If the "exc_info" key in the event dict is either true or a # sys.exc_info() tuple, remove "exc_info" and render the exception # with traceback into the "exception" key. structlog.processors.format_exc_info, # If some value is in bytes, decode it to a Unicode str. structlog.processors.UnicodeDecoder(), # Add callsite parameters. structlog.processors.CallsiteParameterAdder( { structlog.processors.CallsiteParameter.FUNC_NAME, structlog.processors.CallsiteParameter.LINENO, } ), # Render the final event dict as JSON. ] if json_logs: # We rename the `event` key to `message` only in JSON logs, as Datadog looks for the # `message` key but the pretty ConsoleRenderer looks for `event` shared_processors.append(rename_event_key) # Format the exception only for JSON logs, as we want to pretty-print them when # using the ConsoleRenderer shared_processors.append(structlog.processors.format_exc_info) structlog.configure( processors=shared_processors + [ # Prepare event dict for `ProcessorFormatter`. structlog.stdlib.ProcessorFormatter.wrap_for_formatter, ], logger_factory=structlog.stdlib.LoggerFactory(), cache_logger_on_first_use=True, ) log_renderer: structlog.types.Processor if json_logs: log_renderer = structlog.processors.JSONRenderer() else: log_renderer = structlog.dev.ConsoleRenderer() formatter = structlog.stdlib.ProcessorFormatter( # These run ONLY on `logging` entries that do NOT originate within # structlog. foreign_pre_chain=shared_processors, # These run on ALL entries after the pre_chain is done. processors=[ # Remove _record & _from_structlog. structlog.stdlib.ProcessorFormatter.remove_processors_meta, log_renderer, ], ) handler = logging.StreamHandler() # Use OUR `ProcessorFormatter` to format all `logging` entries. handler.setFormatter(formatter) root_logger = logging.getLogger() root_logger.addHandler(handler) root_logger.setLevel(log_level.upper()) for _log in ["uvicorn", "uvicorn.error"]: # Clear the log handlers for uvicorn loggers, and enable propagation # so the messages are caught by our root logger and formatted correctly # by structlog logging.getLogger(_log).handlers.clear() logging.getLogger(_log).propagate = True # Since we re-create the access logs ourselves, to add all information # in the structured log (see the `logging_middleware` in main.py), we clear # the handlers and prevent the logs to propagate to a logger higher up in the # hierarchy (effectively rendering them silent). logging.getLogger("uvicorn.access").handlers.clear() logging.getLogger("uvicorn.access").propagate = False def handle_exception(exc_type, exc_value, exc_traceback): """ Log any uncaught exception instead of letting it be printed by Python (but leave KeyboardInterrupt untouched to allow users to Ctrl+C to stop) See https://stackoverflow.com/a/16993115/3641865 """ if issubclass(exc_type, KeyboardInterrupt): sys.__excepthook__(exc_type, exc_value, exc_traceback) return root_logger.error( "Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback) ) sys.excepthook = handle_exception