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的日志记录模块无缝协作
- 性能:为高吞吐量应用程序优化
- 测试支持:内置测试工具
- 异常处理:增强的异常格式化和渲染
- 自定义处理器:灵活的日志处理管道
最佳实践
- 一次性配置:在应用程序启动时设置structlog配置
- 使用上下文:在请求处理早期绑定相关上下文(request_id、user_id)
- 选择合适的渲染器:开发时使用ConsoleRenderer,生产环境使用JSONRenderer
- 测试日志记录:使用LogCapture进行单元测试日志行为
- 性能优化:缓存日志记录器并为生产环境使用高效的处理器