def configure_logging()

in cortado/utils.py [0:0]

• 53 lines of code
• 9 McCabe index (conditional complexity)

def configure_logging(logging_level: int = logging.DEBUG, as_json: bool = False, root_logger_name: str = "cortado"):
    """Configure logging level and log output format for the root logger.

    By default, logging level is set to DEBUG and logs are printed to stderr as plain text.
    If `as_json` set to `True`, logs are printed as JSON-formatted ECS-valid lines.
    """

    with_colors = False if as_json else True

    # remove all configured log handlers before we set up our own
    for name in logging.root.manager.loggerDict.keys():
        logging.getLogger(name).handlers = []
        logging.getLogger(name).propagate = True

    if as_json:
        attr_processors: list[Any] = []
        additional_processors: list[Any] = [
            structlog.processors.format_exc_info,
            normalise_eventdict_to_ecs,
            # Extend event dict with APM tracing properties
            ecs_logging.StructlogFormatter(),
        ]
    else:
        attr_processors: list[Any] = [
            # Add the name of the logger to event dict
            structlog.stdlib.add_logger_name,
            # Add log level to event dict
            structlog.stdlib.add_log_level,
            # Add a timestamp in ISO 8601 format
            structlog.processors.TimeStamper(fmt="iso"),
        ]
        additional_processors: list[Any] = [
            # If the "stack_info" key in the event dict is true, remove it and
            # render the current stack trace in the "stack" key
            structlog.processors.StackInfoRenderer(),
            # Wrapping is needed in order to use formatter down the stream
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ]

    structlog.configure(
        processors=(
            [
                # If log level is too low, abort pipeline and throw away log entry
                structlog.stdlib.filter_by_level,
            ]  # type: ignore
            + attr_processors
            + [
                # Perform %-style formatting
                structlog.stdlib.PositionalArgumentsFormatter(),
                structlog.processors.CallsiteParameterAdder([CallsiteParameter.FUNC_NAME]),
            ]
            + additional_processors
        ),
        # `wrapper_class` is the bound logger that you get back from get_logger()
        wrapper_class=structlog.stdlib.BoundLogger,
        # `logger_factory` is used to create wrapped loggers that are used for OUTPUT
        logger_factory=structlog.stdlib.LoggerFactory(),
        # Effectively freeze configuration after creating the first bound logger.
        cache_logger_on_first_use=True,
    )

    # Warnings issued by the ``warnings`` module will be redirected to the ``py.warnings`` logger
    logging.captureWarnings(True)

    cortado_root_logger: structlog.stdlib.BoundLogger = structlog.get_logger(root_logger_name)
    cortado_root_logger.setLevel(logging_level)

    # if `as_json` is `True`, event dict will contain only ECS-serialised message
    # that can be sent to `ConsoleRenderer` for output
    renderer = structlog.dev.ConsoleRenderer(colors=with_colors, pad_event=10)
    formatter = structlog.stdlib.ProcessorFormatter(
        foreign_pre_chain=attr_processors,
        processors=[
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            renderer,
        ],
    )

    # https://www.structlog.org/en/stable/standard-library.html#rendering-using-structlog-based-formatters-within-logging
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    handlers = [handler]

    # Reset handlers on a root logger
    logging.getLogger().handlers = handlers  # type: ignore