Pythonのloggingを本気で学ぶ理由
フリーランスとして複数のプロジェクトに関わるようになると、ある瞬間に気づくんですね。本番環境で障害が発生したとき、print()文だけでは何も分からないということに。
数年前、私が担当していたデータパイプライン処理が夜中に止まったんです。理由を調べようと思っても、ログが残っていない。あるのは本番サーバーの悲鳴だけ。その日から、Pythonのloggingモジュールと真摯に向き合うことにしました。
この記事では、単なる「使い方」ではなく、実際のプロジェクトで役に立つログ設計について書きます。バージョンはPython 3.10以上を想定していますが、3.8でもほぼ同じように動きます。
なぜprint()ではダメなのか:アンチパターンから始める
最初のポイントは、アンチパターンを知ることだと思います。
以下のコードを見てください。これは多くの初心者(そして焦っている開発者)が書くパターンです。
def fetch_user_data(user_id):
print(f"ユーザー {user_id} を取得開始")
try:
# データベースアクセス
result = db.query(user_id)
print(f"成功: {result}")
return result
except Exception as e:
print(f"エラーが起きた: {e}")
return None
print()の問題点:
- タイムスタンプがない:いつ起きたのか分からない
- 重要度レベルがない:どれが警告で、どれが致命的なのか区別できない
- ファイルに記録できない:標準出力に流れるだけで、後で見直せない
- 本番環境で消える:ターミナルセッションが終わればログは失われる
- 外部システムと連携できない:ログ管理サービスに送信できない
調べてみたら、Python公式ドキュメントも、ほぼすべての実務プロジェクトも、loggingモジュール推奨してます。理由は明白ですね。
loggingモジュールの基本構造を理解する
ここからはloggingの仕組みを理解します。複雑そうに見えますが、実は4つの部品で成り立っているんです。
Logger, Handler, Formatter, Levelの4層構造
loggingを構成する要素は以下の通り。
- Logger:ログメッセージを受け取るオブジェクト(入口)
- Handler:ログをどこに送るかを決める(ファイル、標準出力など)
- Formatter:ログメッセージの形式を整える(時刻、レベルなど)
- Level:ログの重要度(DEBUG, INFO, WARNING, ERROR, CRITICAL)
図にすると以下のようなイメージです:
Logger(メッセージ入口)→ Filter(フィルタリング) → Handler(出力先決定) → Formatter(形式整形) → 出力
最小限の設定で動かす
まずは最シンプルな例から始めます。
import logging
# 最もシンプルな設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
logger.info("これはINFOレベルのログです")
logger.warning("これはWARNINGレベルのログです")
logger.error("これはERRORレベルのログです")
2024-01-15 14:23:45,123 - __main__ - INFO - これはINFOレベルのログです
2024-01-15 14:23:45,124 - __main__ - WARNING - これはWARNINGレベルのログです
2024-01-15 14:23:45,125 - __main__ - ERROR - これはERRORレベルのログです
basicConfig()は便利ですが、実務では限界があります。本番環境では、ファイル出力とコンソール出力を分けたり、複数のLoggerを使い分けたりする必要が出てくるんですね。
実践的なログ設計パターン:複数Handlerの使い分け
ここからは、実際のプロジェクトで私が使ってる設計です。本番環境を想定して、複数の出力先にログを送り分けるパターンを紹介します。
ファイル出力とコンソール出力を分ける
よくあるシーンとして、開発中はコンソールに見やすく表示して、本番環境ではファイルに詳細情報を記録する、というケースを考えます。
import logging
from logging.handlers import RotatingFileHandler
import os
from datetime import datetime
def setup_logger(name, log_dir='./logs'):
"""複数Handlerを使った実践的なLogger設定"""
# ログディレクトリ作成
os.makedirs(log_dir, exist_ok=True)
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG) # LoggerはDEBUGレベルで受け取る
# フォーマッタ定義
detailed_formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
simple_formatter = logging.Formatter(
'%(levelname)s - %(message)s'
)
# ハンドラ1:ファイル出力(詳細ログ、サイズ制限あり)
file_handler = RotatingFileHandler(
os.path.join(log_dir, f'{name}.log'),
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=5 # 最大5世代保持
)
file_handler.setLevel(logging.DEBUG) # ファイルはすべて記録
file_handler.setFormatter(detailed_formatter)
logger.addHandler(file_handler)
# ハンドラ2:コンソール出力(WARNING以上のみ)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.WARNING) # コンソールはWARNING以上
console_handler.setFormatter(simple_formatter)
logger.addHandler(console_handler)
return logger
# 使用例
logger = setup_logger('myapp')
logger.debug("デバッグ情報(ファイルのみ記録)")
logger.info("通常情報(ファイルのみ記録)")
logger.warning("警告(ファイルとコンソール両方)")
logger.error("エラー(ファイルとコンソール両方)")
WARNING - 警告(ファイルとコンソール両方)
ERROR - エラー(ファイルとコンソール両方)
このパターンの良いところ:
- RotatingFileHandlerで自動的にログファイルをローテーションする。10MBを超えたら新しいファイルに切り替わり、古いファイルは自動削除される
- 開発中はコンソールに警告以上だけ表示されるので邪魔にならない
- 本番環境ではDEBUGレベルまでファイルに記録されるので、トラブルシューティングに役立つ
ログレベルの使い分け方:実務での判断基準
ここ地味に重要です。ログレベルを正しく使い分けられるかで、本番環境での対応速度が変わります。
5段階のレベルと実務での使い分け
| レベル | 数値 | 用途 | 実例 |
|---|---|---|---|
| DEBUG | 10 | 開発中の詳細情報 | 関数の入口・出口、中間変数の値 |
| INFO | 20 | 正常系の重要な出来事 | 処理の開始・完了、ユーザー認証成功 |
| WARNING | 30 | 注意が必要だが処理は続行 | 非推奨APIの使用、レートリミット間近 |
| ERROR | 40 | エラーが発生したが回復可能 | データベース接続失敗(リトライ予定) |
| CRITICAL | 50 | システムが動作不可能 | メモリ枯渇、設定ファイル紛失 |
実際に私が参加したプロジェクトで、この使い分けが甘い開発チームの場合、本番環境のログが「ERROR ERROR ERROR…」だらけになってました。どれが本当に対応が必要なのか判断できない。ストレスですね。
正しいレベル付けの例
import logging
logger = logging.getLogger(__name__)
def process_payment(user_id, amount):
"""支払い処理の例"""
logger.debug(f"支払い処理開始: user_id={user_id}, amount={amount}")
try:
# 残高確認
balance = get_user_balance(user_id)
logger.debug(f"現在の残高: {balance}")
if balance < amount:
logger.warning(f"残高不足: user_id={user_id}, 残高={balance}, 必要額={amount}")
return False
# 決済処理
logger.info(f"決済開始: user_id={user_id}, amount={amount}")
result = payment_gateway.charge(user_id, amount)
logger.info(f"決済完了: user_id={user_id}, transaction_id={result['id']}")
return True
except PaymentGatewayError as e:
# 決済ゲートウェイのエラー:リトライ予定
logger.error(f"決済ゲートウェイエラー: {e}", exc_info=True)
return False
except Exception as e:
# 予期しないエラー:要対応
logger.critical(f"予期しないエラー: {e}", exc_info=True)
raise
exc_info=Trueを使うことで、スタックトレース全体がログに記録されます。本番環境でのトラブルシューティングで非常に役立つんですね。
複数のモジュール・パッケージでのLogger管理
中規模以上のプロジェクトでは、複数のモジュールがあります。モジュール毎にLoggerを分ける設計が重要です。
名前空間を活用したLogger階層化
Pythonのloggingでは、ドット区切りで階層的にLoggerを管理できます。
import logging
# モジュール毎にLoggerを取得
logger_db = logging.getLogger('myapp.database')
logger_api = logging.getLogger('myapp.api')
logger_util = logging.getLogger('myapp.util')
logger_db.debug("データベース操作")
logger_api.info("APIリクエスト受信")
logger_util.warning("ユーティリティ警告")
この設計のメリット:
- モジュール毎にレベルを調整可能:例えば、データベースモジュールはDEBUG、APIモジュールはINFOにする
- ログを出力元で簡単にフィルタリング:特定のモジュールのログだけを見たいときに有効
- 複数パッケージとの連携が簡単:外部ライブラリのログもコントロールできる
プロジェクト全体のログ設定管理
複数モジュールがある場合、全体的なログ設定はconfig.pyやメインモジュールで一元管理するのが良いパターンです。
"""logging_config.py - プロジェクト全体のログ設定"""
import logging
import logging.config
import os
# 環境変数で設定を切り替え
ENV = os.getenv('APP_ENV', 'development')
LOGGING_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s'
},
'simple': {
'format': '%(levelname)s - %(message)s'
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'WARNING' if ENV == 'production' else 'DEBUG',
'formatter': 'simple',
'stream': 'ext://sys.stdout',
},
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'level': 'DEBUG',
'formatter': 'verbose',
'filename': 'logs/app.log',
'maxBytes': 10485760, # 10MB
'backupCount': 5,
},
'error_file': {
'class': 'logging.handlers.RotatingFileHandler',
'level': 'ERROR',
'formatter': 'verbose',
'filename': 'logs/error.log',
'maxBytes': 10485760,
'backupCount': 10,
},
},
'loggers': {
'myapp.database': {
'level': 'DEBUG',
'handlers': ['console', 'file', 'error_file'],
},
'myapp.api': {
'level': 'INFO',
'handlers': ['console', 'file', 'error_file'],
},
'myapp.util': {
'level': 'WARNING',
'handlers': ['console', 'file'],
},
},
'root': {
'level': 'INFO',
'handlers': ['console', 'file', 'error_file'],
},
}
def setup_logging():
"""ログ設定を適用"""
logging.config.dictConfig(LOGGING_CONFIG)
メインモジュールで以下のように呼び出します:
"""main.py"""
from logging_config import setup_logging
import logging
setup_logging()
logger = logging.getLogger('myapp')
if __name__ == '__main__':
logger.info("アプリケーション開始")
# 処理...
この辞書ベースの設定は、JSON設定ファイルとしても保存できるので、本番環境での設定管理が楽になります。
外部ライブラリのログをコントロールする
実務では、FastAPIやSQLAlchemyなど、他のライブラリもログを出力します。これらを適切に制御しないと、本番環境のログがノイズで埋まってしまうんですね。
外部ライブラリのログレベル調整
import logging
from logging_config import setup_logging
setup_logging()
# 外部ライブラリのログレベルを調整
logging.getLogger('sqlalchemy.engine').setLevel(logging.WARNING)
logging.getLogger('urllib3').setLevel(logging.WARNING)
logging.getLogger('fastapi').setLevel(logging.INFO)
logging.getLogger('asyncio').setLevel(logging.WARNING)
logger = logging.getLogger('myapp')
logger.info("設定完了")
実際のプロジェクトでは、この調整がないと本番ログが10GBを超えるなんてことになってます。無駄なストレージ消費と、本当に重要なログを見つけられないという二重のダメージですね。
構造化ログ:JSON形式でのログ出力
本番環境では、ログ管理サービス(Datadog、CloudWatch、ELKスタック等)に送信することが多いです。そのとき、JSON形式の構造化ログがあると便利なんですね。
python-json-loggerを使った構造化ログ
標準のloggingモジュールだけでも頑張れば実装できますが、python-json-loggerライブラリを使うと楽です。
pip install python-json-logger
import logging
from pythonjsonlogger import jsonlogger
from io import StringIO
# JSON形式のハンドラを作成
logstream = StringIO()
json_handler = logging.StreamHandler(logstream)
formatter = jsonlogger.JsonFormatter()
json_handler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(json_handler)
logger.setLevel(logging.INFO)
# ログ出力
logger.info('支払い処理完了', extra={
'user_id': 12345,
'transaction_id': 'txn_abc123',
'amount': 9999,
'duration_ms': 234
})
# JSON形式で出力されている
print(logstream.getvalue())
{"message": "支払い処理完了", "user_id": 12345, "transaction_id": "txn_abc123", "amount": 9999, "duration_ms": 234}
JSON形式のメリット:
- ログサービスが自動的に解析できる:タイムスタンプ、ユーザーID等を個別の値として取得できる
- ダッシュボードで集計できる:「このユーザーのトランザクション数」などの分析が可能
- 検索が効率的:「amount > 10000」という条件検索が可能
本番環境ではほぼ必須だと言えます。
パフォーマンスへの影響と最適化
ログは便利ですが、書き込み頻度が高いと本当のボトルネックになります。実装経験として、以下のポイントが重要です。
ログ出力のパフォーマンス測定
簡単なベンチマークを取ってみました。
import logging
import time
from logging.handlers import RotatingFileHandler
import os
# ログ設定
logger = logging.getLogger('benchmark')
logger.setLevel(logging.DEBUG)
file_handler = RotatingFileHandler('test.log', maxBytes=10*1024*1024, backupCount=5)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# ベンチマーク:10万回のログ出力
start = time.time()
for i in range(100000):
logger.info(f"ログメッセージ {i}")
end = time.time()
print(f"実行時間: {end - start:.2f}秒")
print(f"1メッセージあたり: {(end - start) / 100000 * 1000000:.2f}マイクロ秒")
print(f"ファイルサイズ: {os.path.getsize('test.log') / 1024 / 1024:.2f}MB")
実行時間: 3.45秒
1メッセージあたり: 34.50マイクロ秒
ファイルサイズ: 15.2MB
この結果から分かること:
- 1メッセージ約35マイクロ秒なので、毎秒1万メッセージ程度ならほぼ無視できるレベル
- ただし数万メッセージ/秒では確実にボトルネックになる
高頻度ログ出力の最適化
毎秒数万メッセージが必要な場合は、以下の工夫をします:
import logging
from logging.handlers import QueueHandler, QueueListener
from queue import Queue
from logging.handlers import RotatingFileHandler
# 非同期ログ出力を使用(キューを経由)
log_queue = Queue()
# キューハンドラをメインスレッドに追加
queue_handler = QueueHandler(log_queue)
logger = logging.getLogger('async_app')
logger.setLevel(logging.DEBUG)
logger.addHandler(queue_handler)
# ファイルハンドラをリスナースレッドで実行
file_handler = RotatingFileHandler(
'async_test.log',
maxBytes=10*1024*1024,
backupCount=5
)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
# QueueListenerが別スレッドでログを処理
listener = QueueListener(log_queue, file_handler, respect_handler_level=True)
listener.start()
# 使用例
import time
start = time.time()
for i in range(100000):
logger.info(f"非同期ログ {i}")
end = time.time()
listener.stop()
print(f"実行時間(非同期): {end - start:.2f}秒")
実行時間(非同期): 0.34秒
非同期ログ出力を使うと、約10倍高速になります。ただし、プロセスが突然終了する場合、ログがバッファに残ったまま消える可能性があるので、適切なシャットダウン処理が必要ですね。
よくある落とし穴と対策
実務で何度も見かけたミスをまとめます。
basicConfig()の重複呼び出し
これ、本当に多い落とし穴です。
"""NG例"""
import logging
# main.py
logging.basicConfig(level=logging.INFO)
# util.py
logging.basicConfig(level=logging.DEBUG) # 反映されない!
basicConfig()は最初の呼び出し時だけ有効です。2番目の呼び出しは無視されます。対策は、最初から辞書設定を使うこと、またはgetLogger()で直接Handlerを追加することですね。
例外情報が記録されない
"""NG例"""
try:
1 / 0
except Exception as e:
logger.error(f"エラー: {e}") # スタックトレースが記録されない
"""OK例"""
try:
1 / 0
except Exception as e:
logger.error(f"エラー: {e}", exc_info=True) # スタックトレース付き
exc_info=Trueまたはlogger.exception()を使いましょう。
ログメッセージの文字列フォーマットのパフォーマンス
"""NG例"""
for i in range(1000000):
logger.debug(f"値: {expensive_function()}") # DEBUGレベルを出力しない設定でも、expensive_function()は呼ばれる
"""OK例"""
for i in range(1000000):
logger.debug("値: %s", expensive_function()) # または遅延フォーマット
いや、もっと正確には:
"""最も効率的"""
if logger.isEnabledFor(logging.DEBUG):
logger.debug("値: %s", expensive_function())
ログレベルフィルタリング前にf-stringの処理が走らないようにすることが重要です。
テスト時のログ検証
単体テストでログが正しく出力されているか検証することも実務では重要ですね。
"""test_logging.py"""
import logging
import unittest
from io import StringIO
class TestLogging(unittest.TestCase):
def setUp(self):
# ログ出力をキャプチャ
self.log_stream = StringIO()
self.handler = logging.StreamHandler(self.log_stream)
self.handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(levelname)s - %(message)s')
self.handler.setFormatter(formatter)
self.logger = logging.getLogger('test')
self.logger.addHandler(self.handler)
self.logger.setLevel(logging.DEBUG)
def tearDown(self):
self.logger.removeHandler(self.handler)
self.handler.close()
def test_log_output(self):
self.logger.info("テストメッセージ")
log_output = self.log_stream.getvalue()
self.assertIn("テストメッセージ", log_output)
self.assertIn("INFO", log_output)
if __name__ == '__main__':
unittest.main()
ユニットテストで「このエラーメッセージは出力されるべき」という検証ができると、リファクタリング時の安心感が全く違いますね。
本番環境への展開を想定した設定例
最後に、実際のプロジェクトで使ってる設定を丸ごと紹介します。FastAPIアプリケーションを想定しています。
"""production_logging_config.py - 本番環境用ログ設定"""
import logging
import logging.config
import os
from pythonjsonlogger import jsonlogger
LOG_DIR = os.getenv('LOG_DIR', '/var/log/myapp')
ENV = os.getenv('APP_ENV', 'production')
os.makedirs(LOG_DIR, exist_ok=True)
LOGGING_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'json': {
'()': 'pythonjsonlogger.jsonlogger.JsonFormatter',
'format': '%(asctime)s %(name)s %(levelname)s %(message)s',
},
'verbose': {
'format': '%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(funcName)s - %(message)s',
'datefmt': '%Y-%m-%d %H:%M:%S',
},
},
'handlers': {
'app_file': {
'class': 'logging.handlers.RotatingFileHandler',
'level': 'INFO',
'formatter': 'json',
'filename': os.path.join(LOG_DIR, 'app.log'),
'maxBytes': 100 * 1024 * 1024, # 100MB
'backupCount': 10,
'encoding': 'utf-8',
},
'error_file': {
'class': 'logging.handlers.RotatingFileHandler',
'level': 'ERROR',
'formatter': 'json',
'filename': os.path.join(LOG_DIR, 'error.log'),
'maxBytes': 100 * 1024 * 1024,
'backupCount': 20,
'encoding': 'utf-8',
},
'console': {
'class': 'logging.StreamHandler',
'level': 'WARNING',
'formatter': 'verbose',
},
},
'loggers': {
'myapp': {
'level': 'DEBUG',
'handlers': ['app_file', 'error_file', 'console'],
'propagate': False,
},
'sqlalchemy': {
'level': 'WARNING',
'handlers': ['app_file'],
'propagate': False,
},
'asyncio': {
'level': 'WARNING',
'handlers': ['app_file'],
'propagate': False,
},
},
'root': {
'level': 'INFO',
'handlers': ['app_file', 'error_file'],
},
}
def setup_logging():
logging.config.dictConfig(LOGGING_CONFIG)
これを本番環境で使用する場合:
"""main.py"""
from fastapi import FastAPI
from production_logging_config import setup_logging
import logging
setup_logging()
logger = logging.getLogger('myapp')
app = FastAPI()
@app.on_event('startup')
async def startup():
logger.info("アプリケーション起動", extra={
'version': '1.0.0',
'environment': 'production'
})
@app.on_event('shutdown')
async def shutdown():
logger.info("アプリケーションシャットダウン")
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host='0.0.0.0', port=8000)
この設定で:
- すべてのINFO以上のログがJSON形式で
app.logに記録される - ERROR以上は別途
error.logに記録される(トラブルシューティング時の検索が楽) - コンソールにはWARNING以上だけが見易い形式で表示される
- ログサーバーにJSONを送信すれば、ダッシュボード化も可能
まとめ
Pythonのloggingモジュールは、初心者には複雑に見えるかもしれません。しかし、Logger、Handler、Formatter、Levelの4つの要素を理解すれば、本当に柔軟で強力です。
実務で大切なポイント:
print()ではなくloggingを使う。タイムスタンプと重要度レベルがあるだけで、本番環境での対応が段違い- 複数のHandlerを使い分ける。ファイル出力とコンソール出力を分けることで、開発と本番環境の両立が可能
- ログレベルを正しく使い分ける。「WARNING = 注意が必要だが回復可能」という基準を持つ
- 複数モジュールがある場合は、名前空間で階層化管理する
- 本番環境ではJSON形式の構造化ログにする。ログ管理サービスとの連携が圧倒的に楽
- 高頻度ログはQueueHandlerで非同期化。メインスレッドのブロッキングを避ける
- 外部ライブラリのログレベルは必ず調整。ノイズで本当のログが見えなくなる
よくある質問(FAQ)
basicConfig()と辞書設定の使い分けは?
小さなスクリプト(100行以下)ならbasicConfig()で十分です。でも複数モジュール、複数出力先を持つプロジェクトなら、最初から辞書設定を使いましょう。後で変更する手間が大きいので。
ログファイルのサイズが巨大になる場合の対策は?
必ずRotatingFileHandlerを使ってください。maxBytesとbackupCountで自動的にローテーションされます。無制限に書き込んでいると、ディスクが満杯になります。実務で経験済みです(苦い思い出)。
複数プロセスから同じログファイルに書き込む場合は?
標準のRotatingFileHandlerは、複数プロセス時にロック機構がないため、ログが破損する可能性があります。logging.handlers.QueueHandlerとQueueListenerを組み合わせて、シングルスレッドでログ書き込みするか、ConcurrentLogHandler(サードパーティ)を使用してください。
ログを外部ツール(DatadogやCloudWatch)に送信するには?
JSON形式の構造化ログを出力しておけば、Logstash、Fluentd、CloudWatch Agentなどが自動的に解析・転送してくれます。ログレベルとメッセージだけでなく、extraパラメータに構造化データを含めることで、検索精度が高まります。
本番環境でログレベルをDEBUGのままにしても大丈夫?
ディスク容量が許すなら、悪くない選択肢です。トラブルシューティング時に詳細情報があると助かります。ただし、毎秒大量のログが出ている場合は、パフォーマンス影響を考えてWARNING程度に上げるのが無難です。本番環境では常にモニタリングしておきましょう。

