结构化日志记录库Skill "structlog"

Structlog是一个强大的Python结构化日志记录库,专为现代应用程序设计。它提供上下文感知的日志记录、多种输出格式(JSON/控制台/logfmt)、高性能处理管道以及与Python标准日志模块的无缝集成。主要功能包括:结构化日志输出、上下文绑定、自定义处理器、异常追踪增强、测试工具支持。适用于后端开发、微服务架构、DevOps监控等场景,帮助开发者实现可维护、可搜索、机器可读的日志系统。关键词:Python日志记录、结构化日志、JSON日志、上下文日志、日志处理器、开发运维、后端开发、微服务监控。

后端开发 0 次安装 0 次浏览 更新于 3/2/2026

name: “structlog” description: “为Python应用程序提供结构化日志记录,支持上下文和强大的处理器” when_to_use: “当您需要为Python应用程序提供带上下文的结构化日志记录、JSON输出或增强的日志记录功能时”

Structlog 技能

快速开始

import structlog

# 基本用法
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

# 配置JSON输出
structlog.configure(
    processors=[structlog.processors.JSONRenderer()]
)

log = structlog.get_logger()
log.info("处理请求", request_id="req-123", user_id=456)
# 输出: {"event": "处理请求", "request_id": "req-123", "user_id": 456}

标准库集成

import logging
import structlog

# 配置标准日志记录
logging.basicConfig(
    format="%(message)s",
    stream=sys.stdout,
    level=logging.INFO
)

# 配置structlog以使用标准库
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()

# 绑定在多个日志调用中持久存在的上下文
request_log = log.bind(request_id="req-789", user="alice")
request_log.info("处理开始")
request_log.info("数据库查询已执行", query="SELECT * FROM users")
request_log.info("处理完成")

# 输出在所有日志条目中都包含request_id和user

自定义处理器

import time

def add_custom_context(logger, log_method, event_dict):
    """为每个日志条目添加自定义上下文"""
    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("发生除法错误")

使用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("测试消息", key="value")

    assert len(log_output.entries) == 1
    assert log_output.entries[0]["event"] == "测试消息"
    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的日志记录模块无缝协作
  • 性能:为高吞吐量应用程序优化
  • 测试支持:内置测试工具
  • 异常处理:增强的异常格式化和渲染
  • 自定义处理器:灵活的日志处理管道

最佳实践

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