Pythonのロギング：本番環境向けの設定

1. 概要

print()は小さなスクリプトを書く際には十分です。ただし本番環境では異なります。ログレベルでフィルタリングし、ログを集約システムに統合し、複数のログ行を通じてリクエストを追跡し、ログの書き込みがアプリケーションを遅くしてはいけません。

この記事はPythonの標準loggingモジュールを本番環境向けに設定する方法を説明します。読み終わったら、モジュール別logger、dictConfig、JSON形式ログ、自動的に付与されるrequest_id、QueueHandlerによる非同期ログ出力、およびpytestのcaplogを使ったログテストなど、完全な設定を構築できるようになります。すべての例はPython 3.12を使用し、標準ライブラリのみが必要です。テストの部分のみpytestが必要です。

2. printとbasicConfigが不十分な理由

注文を処理するサービスを「とりあえず完成させる」というスタイルで書いたコードを見てみましょう：

import logging

logging.basicConfig(level=logging.INFO)

def place_order(order_id: str, amount: float) -> None:
    print(f"Placing order {order_id}")
    logging.info(f"Order {order_id} amount={amount}")

place_order("ORD-1001", 250_000)

Placing order ORD-1001
INFO:root:Order ORD-1001 amount=250000

このコードには3つの問題があります：

• printはレベル、タイムスタンプがなく、設定で無効化できません。
• logging.info(...)はルートロガーに書き込みます。ログがどのモジュールから来たのか分かりませんし、各部分のレベルを個別に調整できません。
• basicConfigは最初の呼び出しでのみ効果があります。root loggerがハンドラーをまだ持たない場合です。別のライブラリが最初に呼び出した場合、あなたの設定は警告なしに無視されます。

3. getLogger(__name__)によるモジュール別logger

最初のルール：各モジュールは独自のloggerを作成し、__name__で名前を付けます。

# shop/orders.py
import logging

logger = logging.getLogger(__name__)  # logger名: "shop.orders"

class OrderService:
    def place_order(self, order_id: str, quantity: int) -> None:
        logger.info("Placing order %s with quantity %d", order_id, quantity)
        if quantity > 50:
            logger.warning("Large order %s: quantity=%d", order_id, quantity)

logger名はドット区切りの階層構造を形成します。shop.ordersはshopの子で、shopはrootの子です。レコードは子loggerから親loggerに上がります（これをpropagationといいます）。したがって、rootまたはshopレベルでハンドラーをアタッチし、各ブランチのレベルを個別に調整するだけで十分です。たとえば、shop.paymentsをDEBUGに設定して、残りの部分をINFOに保つことができます。

パラメータの渡し方にも注意してください：f文字列ではなく%sを使用し、引数として渡します。レベルが無効の場合、文字列はフォーマットされないため、決して出力されないログのコストを無駄にしません。さらに、ログ集約ツールは同じメッセージテンプレートのログをグループ化できます。

注意： ライブラリ（他のユーザーが使用するために発行するパッケージ）では、ハンドラーを設定しないでください。getLogger(__name__)を呼び出すだけで、アプリケーションに決定を任せます。必要に応じて、logging.getLogger("your_lib").addHandler(logging.NullHandler())を追加します。

4. dictConfigによる集約設定

本番環境では、すべての設定を1つの場所に集め、アプリケーション起動時に1度だけ呼び出すべきです。logging.config.dictConfigはこの目的に適しています。

# shop/logging_config.py
import logging.config
import os

def setup_logging() -> None:
    level = os.getenv("LOG_LEVEL", "INFO")
    logging.config.dictConfig({
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "plain": {
                "format": "%(asctime)s %(levelname)-8s %(name)s - %(message)s",
            },
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
                "formatter": "plain",
            },
        },
        "loggers": {
            "shop": {"level": level},
            "urllib3": {"level": "WARNING"},
        },
        "root": {"level": "WARNING", "handlers": ["console"]},
    })

設定内の主要なポイント：

• "disable_existing_loggers": Falseは実際には必須です。デフォルト値のTrueは、dictConfig実行前に作成されたすべてのloggerを無効化します。これはあなたが各モジュールの先頭で import したloggerも含みます。結果としてログが消えますが、エラーは出ません。
• ハンドラーはrootにのみアタッチされます。logger shopはレベルのみを設定し、レコードはrootまで伝播して出力されます。
• rootはWARNINGに設定され、サードパーティライブラリからのうるさいINFOログを遮断します。shopブランチのコードは引き続きINFOでログします。
• レベルはLOG_LEVEL環境変数から読み込まれます。問題を調査する際、コードを変更せずにDEBUGを有効にできます。

エントリーポイントで関数を呼び出します：

# main.py
from shop.logging_config import setup_logging
from shop.orders import OrderService

setup_logging()
OrderService().place_order("ORD-1001", quantity=80)

2026-10-18 09:15:02,114 INFO     shop.orders - Placing order ORD-1001 with quantity 80
2026-10-18 09:15:02,114 WARNING  shop.orders - Large order ORD-1001: quantity=80

stdoutへの出力は、コンテナ内で実行するときの正しい選択です。DockerまたはKubernetesはこのストリームからログを収集するため、ファイルを自分で管理したりログローテーションを心配する必要はありません。

5. JSON形式の構造化ログ

LokiやElasticsearch、CloudWatchなどのログ集約システムは、自由形式のテキストよりもJSONを検索する方がはるかに優れています。外部ライブラリなしでフォーマッターを自分で書くことができます：

# shop/json_formatter.py
import json
import logging
from datetime import datetime, timezone

_RESERVED = set(vars(logging.LogRecord("", 0, "", 0, "", None, None))) | {"message", "asctime"}

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        for key, value in record.__dict__.items():
            if key not in _RESERVED and not key.startswith("_"):
                payload[key] = value
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False, default=str)

仕組み：

• _RESERVEDはLogRecordのデフォルト属性（pathname、lineno、argsなど）を含みます。
• 他のすべての属性はカスタムフィールドと見なされ、JSONに追加されます。これらのフィールドは通常、extra=パラメータまたはフィルタから来ます（セクション6を参照）。
• default=strは、DecimalやUUIDなどのタイプに出会ったときにjson.dumpsがエラーにならないようにします。

dictConfigでフォーマッターを宣言し、特殊キー"()"を使用します：

"formatters": {
    "json": {"()": "shop.json_formatter.JsonFormatter"},
},
"handlers": {
    "console": {
        "class": "logging.StreamHandler",
        "stream": "ext://sys.stdout",
        "formatter": "json",
    },
},

extraでビジネスフィールドを追加します：

logger.info("Payment captured", extra={"order_id": "ORD-1001", "amount": 250000})

{"ts": "2026-10-18T02:15:02.114+00:00", "level": "INFO", "logger": "shop.payments", "message": "Payment captured", "order_id": "ORD-1001", "amount": 250000}

注意： extra内のキーは、LogRecordの既存の属性名と重複してはいけません。たとえば、extra={"name": "..."} または extra={"message": "..."} は KeyError: "Attempt to overwrite 'name' in LogRecord" をスローします。

6. contextvarsを使用した自動request_id付与

毎回のログ呼び出しでextra={"request_id": ...}を渡すのは冗長で忘れやすいです。より良い方法は、request_idをContextVarに保存し、Filterでそれをすべてのレコードに自動的に付与することです。

# shop/request_context.py
import logging
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()
        return True  # レコードをブロックしない、データを補加するのみ

ContextVarはスレッドとasyncioの両方で安全です。各タスクまたはリクエストは独自の値を保持し、あるリクエストが別のリクエストのIDを「借りる」ことはありません。

フィルターをloggerではなく、ハンドラーにアタッチします。loggerにアタッチされたフィルターは、そのloggerから直接作成されたレコードに対してのみ実行され、子loggerから伝播してくるレコードには実行されません。

"filters": {
    "request_id": {"()": "shop.request_context.RequestIdFilter"},
},
"handlers": {
    "console": {
        "class": "logging.StreamHandler",
        "stream": "ext://sys.stdout",
        "formatter": "json",
        "filters": ["request_id"],
    },
},

各リクエストの開始時に値を設定します。例えば、ミドルウェアで：

import uuid
from shop.request_context import request_id_var

def handle_request(order_id: str) -> None:
    token = request_id_var.set(uuid.uuid4().hex[:12])
    try:
        OrderService().place_order(order_id, quantity=3)
    finally:
        request_id_var.reset(token)

{"ts": "...", "level": "INFO", "logger": "shop.orders", "message": "Placing order ORD-1002 with quantity 3", "request_id": "9f2c1ab07d3e"}

これでrequest_idでフィルタリングするだけで、リクエストの全体的な流れが見えます。

7. 例外ログの正しい記録

logger = logging.getLogger("shop.payments")

class PaymentService:
    def charge(self, order_id: str, amount: int) -> bool:
        try:
            raise ConnectionError("gateway timeout")
        except ConnectionError:
            logger.exception("Charge failed for order %s", order_id)
            return False

logger.exception(...)はERRORレベルで記録し、スタックトレースを自動的に添付します。これはexceptブロック内でのみ呼び出します。別のレベルでスタックトレースと共に記録したい場合は、logger.warning("...", exc_info=True)を使用します。

よくある2つのミス：

• logger.error(f"Charge failed: {e}")はスタックトレースを失います。調査する際、「gateway timeout」という1行だけが見え、それがどこで発生したかは分かりません。
• 各レイヤーでログと再スローの両方を行うこと。同じ例外が5回ログに出現します。例外処理を行うレイヤー、つまり再スローしないことに決めた場所でログしてください。

8. ログがアプリケーションをブロックしない：QueueHandler

StreamHandlerがstdoutに書き込むのは通常十分な速度です。しかし、syslogやHTTP、SMTPなどを経由するハンドラーやディスクへの書き込みが遅い場合、リクエストを処理しているスレッド自体がブロックされます。これを回避するために、レコードをキューに入れ、バックグラウンドスレッドが実際の出力先に書き込みます：

import atexit
import logging
import logging.handlers
import queue

def setup_async_logging(target: logging.Handler) -> None:
    log_queue: queue.Queue = queue.Queue(-1)
    queue_handler = logging.handlers.QueueHandler(log_queue)

    listener = logging.handlers.QueueListener(
        log_queue, target, respect_handler_level=True
    )
    listener.start()
    atexit.register(listener.stop)  # 終了時に残りをフラッシュ

    root = logging.getLogger()
    root.handlers = [queue_handler]

処理フロー：

• アプリケーションスレッドは、レコードをqueueにputすることだけを行います。この操作はほぼ即座です。
• QueueListenerは独立したスレッドで実行され、キューからレコードを取り出し、target（例えばSysLogHandler）に転送します。
• atexit.register(listener.stop)は、プロセス終了前に最後のログが書き込まれることを保証します。

注意： RequestIdFilterなどのフィルターは、handler送信先ではなく、QueueHandlerにアタッチする必要があります。handler送信先はリスナースレッドで実行され、そのスレッドのContextVarは元のリクエストの値を持たないからです。

9. pytestのcaplogでログをテストする

ビジネスロジックの警告、例えば大量注文は、テストされるべき動作です。dev依存関係にpytestを追加します：

pip install pytest

# tests/test_orders.py
import logging
from shop.orders import OrderService

def test_large_order_logs_warning(caplog):
    with caplog.at_level(logging.INFO, logger="shop.orders"):
        OrderService().place_order("ORD-2001", quantity=80)

    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
    assert len(warnings) == 1
    assert warnings[0].name == "shop.orders"
    assert warnings[0].getMessage() == "Large order ORD-2001: quantity=80"

def test_small_order_does_not_warn(caplog):
    with caplog.at_level(logging.INFO, logger="shop.orders"):
        OrderService().place_order("ORD-2002", quantity=2)

    assert all(r.levelno < logging.WARNING for r in caplog.records)

tests/test_orders.py ..                                  [100%]
2 passed in 0.03s

caplog.at_levelはwithブロック内で一時的にlogger shop.ordersのレベルを下げます。caplog.recordsは元のLogRecordを含むため、レベル、loggerの名前、およびextraフィールド（存在する場合）をチェックできます。

注意： caplogはrootロガーにハンドラーをアタッチします。設定でlogger shopに対して"propagate": Falseを設定した場合、テストはレコードをキャッチしません。

10. よくあるミスと避けるべきこと

• モジュールレベルでのロギング設定（import時にbasicConfigまたはdictConfigを呼び出す）。エントリーポイントで1回だけ設定してください。
• ハンドラーが2回アタッチされる。例えば、propagateがTrueのままshopとrootの両方にアタッチされている場合。各ログ行が2回出力されます。ハンドラーは1つのレイヤーにのみアタッチすべきです。
• 機密データを記録する。パスワード、トークン、カード番号です。リクエストオブジェクト全体をログしないでください。必要な場合は、これらのフィールドをマスクするFilterを書いてください。
• 複数プロセスアプリケーションでRotatingFileHandlerを使用する（複数workerを持つGunicorn）。複数のプロセスが同じファイルをローテーションすると、ログが失われます。stdoutに出力するか、各workerから単一の書き込みプロセスにQueueHandler経由でプッシュしてください。
• ホットループでINFOを過度にログしたい。ストレージコストと雑音が非常に速く増加します。このようなログはDEBUGに置くべきです。
• 開発環境でJSONを使用したい。JSONは読みにくいため、環境変数に応じてフォーマッターを選択してください。開発環境ではplain、本番環境ではjson。

11. まとめ

本番環境向けのロギング設定は複雑ではなく、一貫性が必要です。すべてのモジュールでgetLogger(__name__)を使用します。dictConfigの単一呼び出しで設定を集約し、disable_existing_loggers: Falseを設定します。JSON形式をstdoutに出力し、ContextVarとFilter経由でrequest_idを付与し、logger.exceptionを使ってスタックトレースを保持します。handler送信先が遅い場合は、QueueHandlerとQueueListenerに切り替えます。

最後に、ログをシステムの動作の一部と見なし、caplogでテストしてください。午前2時に問題が発生した時、構造化されたこれらのログ行があれば、原因を非常に迅速に特定できます。