structlog

用途

structlog 是一个面向 Python 应用的结构化日志库。它以“事件 + 键值对”的形式记录日志，使日志内容可被机器解析，同时提供上下文绑定和可扩展的处理器管线，便于在应用中统一管理日志输出格式与附加信息。

主要能力

结构化日志：以字典形式记录事件与上下文键值，示例：log.info("hello, %s!", "world", key="value", more_than_strings=[1, 2, 3])。
多种输出格式：支持控制台、JSON、logfmt 以及自定义渲染器。
上下文绑定：通过 bind 绑定在后续多次日志调用中持续存在的上下文，例如：

log = structlog.get_logger()
request_log = log.bind(request_id="req-789", user="alice")
request_log.info("Processing started")

标准库集成：可与 Python 标准 logging 模块配合使用，包括 structlog.stdlib.LoggerFactory、structlog.stdlib.BoundLogger、ProcessorFormatter 等。
自定义处理器：处理器以函数形式接收 logger, log_method, event_dict 并返回修改后的字典，可插入处理链。
异常处理：提供 structlog.processors.dict_tracebacks、format_exc_info 等，配合 log.exception() 记录异常。
测试支持：内置 structlog.testing.LogCapture，可在 pytest 等测试框架中断言日志条目内容。
性能优化：可通过缓存 logger（cache_logger_on_first_use=True）与高效处理器配置提升吞吐。

使用方式

基础配置：

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.StackInfoRenderer(),
        structlog.dev.set_exc_info,
        structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
        structlog.dev.ConsoleRenderer()
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.NOTSET),
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=False
)

JSON 输出：

import structlog

structlog.configure(
    processors=[structlog.processors.JSONRenderer()]
)

log = structlog.get_logger()
log.info("Processing request", request_id="req-123", user_id=456)
# Output: {"event": "Processing request", "request_id": "req-123", "user_id": 456}

标准库集成：

import logging
import structlog

logging.basicConfig(
    format="%(message)s",
    stream=sys.stdout,
    level=logging.INFO
)

structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.stdlib.render_to_log_kwargs,
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

自定义处理器：

import time

def add_custom_context(logger, log_method, event_dict):
    """Add custom context to every log entry"""
    event_dict["custom_field"] = "custom_value"
    event_dict["timestamp"] = time.time()
    return event_dict

structlog.configure(
    processors=[
        add_custom_context,
        structlog.processors.JSONRenderer()
    ]
)

异常处理：

import structlog

structlog.configure(
    processors=[
        structlog.processors.dict_tracebacks,
        structlog.processors.JSONRenderer(),
    ],
)

log = structlog.get_logger()

try:
    1 / 0
except ZeroDivisionError:
    log.exception("Division error occurred")

测试：

import pytest
import structlog
from structlog.testing import LogCapture

@pytest.fixture
def log_output():
    return LogCapture()

@pytest.fixture(autouse=True)
def configure_structlog(log_output):
    structlog.configure(
        processors=[log_output]
    )

def test_logging(log_output):
    log = structlog.get_logger()
    log.info("test message", key="value")

    assert len(log_output.entries) == 1
    assert log_output.entries[0]["event"] == "test message"
    assert log_output.entries[0]["key"] == "value"

性能优化配置：

import structlog

structlog.configure(
    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(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

高级控制台输出（含文件与控制台双通道）：

import logging.config
import structlog

timestamper = structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S")
pre_chain = [
    structlog.stdlib.add_log_level,
    structlog.stdlib.ExtraAdder(),
    timestamper,
]

logging.config.dictConfig({
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "plain": {
            "()": structlog.stdlib.ProcessorFormatter,
            "processors": [
                structlog.stdlib.ProcessorFormatter.remove_processors_meta,
                structlog.dev.ConsoleRenderer(colors=False),
            ],
            "foreign_pre_chain": pre_chain,
        },
        "colored": {
            "()": structlog.stdlib.ProcessorFormatter,
            "processors": [
                structlog.stdlib.ProcessorFormatter.remove_processors_meta,
                structlog.dev.ConsoleRenderer(colors=True),
            ],
            "foreign_pre_chain": pre_chain,
        },
    },
    "handlers": {
        "default": {
            "level": "DEBUG",
            "class": "logging.StreamHandler",
            "formatter": "colored",
        },
        "file": {
            "level": "DEBUG",
            "class": "logging.handlers.WatchedFileHandler",
            "filename": "app.log",
            "formatter": "plain",
        },
    },
    "loggers": {
        "": {
            "handlers": ["default", "file"],
            "level": "DEBUG",
        }
    }
})

关键特性

结构化日志：以字典记录事件与上下文。
多输出格式：控制台、JSON、logfmt 及自定义渲染器。
上下文绑定：跨日志调用保持上下文。
标准库集成：与 Python logging 模块协同工作。
性能：面向高吞吐应用进行了优化。
测试支持：内置测试工具。
异常处理：增强的异常格式化与渲染。
自定义处理器：灵活的日志处理管线。

最佳实践

只配置一次：在应用启动时完成 structlog 配置。
使用上下文：在请求处理早期绑定相关上下文（request_id、user_id）。
选择合适渲染器：开发用 ConsoleRenderer，生产用 JSONRenderer。
测试日志：使用 LogCapture 对日志行为做单元测试。
性能：缓存 logger，并在生产环境使用高效处理器。

适用场景与限制

适用于需要结构化、可机器解析日志的 Python 应用，尤其是服务端、请求处理链路中需要携带 request_id、user_id 等上下文的场景。开发阶段适合控制台彩色输出，生产环境适合 JSON 输出以便采集与检索。

关于许可协议、安装方式、版本要求、收费模式、支持的 Python 版本范围等具体信息，资料中暂未提供。