Configure structlog and stdlib logging.
Structlog is wired through stdlib so that both first-party loggers (structlog.get_logger()) and third-party stdlib loggers share the same handler chain. The file handler always emits JSON; the console handler emits colourised output by default or JSON when use_json_formatter is True.
Source code in src/cli_app/utils/log.py
| def setup_logging(config: LogConfig | None = None) -> None:
"""Configure structlog and stdlib logging.
Structlog is wired through stdlib so that both first-party loggers
(``structlog.get_logger()``) and third-party stdlib loggers share the
same handler chain. The file handler always emits JSON; the console
handler emits colourised output by default or JSON when
``use_json_formatter`` is ``True``.
"""
if config is None:
config = LogConfig()
logs_dir_path = config.dir.resolve()
logs_dir_path.mkdir(parents=True, exist_ok=True)
log_file_path = logs_dir_path / config.file_name
# Processors applied to every log record (structlog and foreign stdlib).
shared_processors: list[structlog.types.Processor] = [
structlog.contextvars.merge_contextvars,
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
]
console_renderer: structlog.types.Processor = (
structlog.processors.JSONRenderer()
if config.use_json_formatter
else structlog.dev.ConsoleRenderer()
)
# File handler always writes JSON for structured log analysis.
file_formatter = structlog.stdlib.ProcessorFormatter(
processors=[
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
structlog.processors.JSONRenderer(),
],
foreign_pre_chain=shared_processors,
)
console_formatter = structlog.stdlib.ProcessorFormatter(
processors=[
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
console_renderer,
],
foreign_pre_chain=shared_processors,
)
root_logger = logging.getLogger()
root_logger.setLevel(config.level)
existing_types = {type(h) for h in root_logger.handlers}
if logging.handlers.RotatingFileHandler not in existing_types:
fh = logging.handlers.RotatingFileHandler(
filename=log_file_path,
maxBytes=config.file_max_bytes,
backupCount=config.file_backup_count,
)
fh.setLevel(config.file_level or config.level)
fh.setFormatter(file_formatter)
root_logger.addHandler(fh)
if logging.StreamHandler not in existing_types:
sh = logging.StreamHandler(sys.stderr)
sh.setLevel(config.console_level or config.level)
sh.setFormatter(console_formatter)
root_logger.addHandler(sh)
structlog.configure(
processors=[*shared_processors, structlog.stdlib.ProcessorFormatter.wrap_for_formatter],
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
|