structlog

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Structlog Skill

Structlog 使用指南

Quick Start

快速开始

python
import structlog
python
import structlog

Basic usage

Basic usage

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

Common Patterns

常见用法

Basic Configuration

基础配置

python
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
)
python
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 Logging

JSON 日志输出

python
import structlog
python
import structlog

Configure for JSON output

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

Output: {"event": "Processing request", "request_id": "req-123", "user_id": 456}

undefined
undefined

Standard Library Integration

与标准库集成

python
import logging
import structlog
python
import logging
import structlog

Configure standard logging

Configure standard logging

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

Configure structlog to use standard library

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, )
undefined
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, )
undefined

Context Binding

上下文绑定

python
import structlog

log = structlog.get_logger()
python
import structlog

log = structlog.get_logger()

Bind context that persists across log calls

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

Output includes request_id and user in all log entries

undefined
undefined

Custom Processors

自定义处理器

python
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()
    ]
)
python
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()
    ]
)

Exception Handling

异常处理

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

Testing with Structlog

Structlog 测试

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

Performance-Optimized Configuration

性能优化配置

python
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,
)
python
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,
)

Advanced Console Output

高级控制台输出

python
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",
        }
    }
})
python
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",
        }
    }
})

Key Features

核心特性

  • Structured Logging: Log events as dictionaries with context
  • Multiple Output Formats: Console, JSON, logfmt, and custom renderers
  • Context Binding: Persistent context across log calls
  • Standard Library Integration: Works seamlessly with Python's logging module
  • Performance: Optimized for high-throughput applications
  • Testing Support: Built-in testing utilities
  • Exception Handling: Enhanced exception formatting and rendering
  • Custom Processors: Flexible pipeline for log processing
  • 结构化日志:将日志事件以带上下文的字典形式记录
  • 多输出格式:支持控制台、JSON、logfmt及自定义渲染器
  • 上下文绑定:日志调用间可持久化上下文
  • 标准库集成:与Python logging模块无缝协作
  • 高性能:针对高吞吐量应用优化
  • 测试支持:内置测试工具
  • 异常处理:增强的异常格式化与渲染
  • 自定义处理器:灵活的日志处理流水线

Best Practices

最佳实践

  1. Configure Once: Set up structlog configuration at application startup
  2. Use Context: Bind relevant context (request_id, user_id) early in request handling
  3. Choose Right Renderer: Use ConsoleRenderer for development, JSONRenderer for production
  4. Test Logging: Use LogCapture for unit testing logging behavior
  5. Performance: Cache loggers and use efficient processors for production
  1. 统一配置:在应用启动时完成structlog配置
  2. 善用上下文:在请求处理初期绑定相关上下文(如request_id、user_id)
  3. 选择合适的渲染器:开发环境使用ConsoleRenderer,生产环境使用JSONRenderer
  4. 测试日志行为:使用LogCapture进行日志行为的单元测试
  5. 性能优化:生产环境中缓存日志实例并使用高效处理器