structlog

Structlog 技能

快速开始

import structlog

# Basic usage
log = structlog.get_logger()
log.info("hello, %s!", "world", key="value", more_than_strings=[1, 2, 3])

常见模式

基本配置

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

# Configure for JSON output
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

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

# Configure structlog to use standard library
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 structlog

log = structlog.get_logger()

# Bind context that persists across log calls
request_log = log.bind(request_id="req-789", user="alice")
request_log.info("Processing started")
request_log.info("Database query executed", query="SELECT * FROM users")
request_log.info("Processing completed")

# Output includes request_id and user in all log entries

自定义处理器

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")

使用 Structlog 进行测试

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 模块无缝协作
• 性能：针对高吞吐量应用进行了优化
• 测试支持：内置测试工具
• 异常处理：增强异常格式化和渲染
• 自定义处理器：灵活的日志处理管道

最佳实践

1. 只配置一次：在应用启动时设置 structlog 配置
2. 使用上下文：在请求处理早期绑定相关上下文（request_id、user_id）
3. 选择合适的渲染器：开发时使用 ConsoleRenderer，生产环境使用 JSONRenderer
4. 测试日志记录：使用 LogCapture 对日志行为进行单元测试
5. 性能：在生产环境中缓存日志记录器并使用高效处理器