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. パフォーマンス：本番環境ではロガーをキャッシュし、効率的なプロセッサーを使用します