logging (Standard Library)
Python標準ライブラリの強力なロギングフレームワーク。設定可能性と拡張性に優れ、複雑な企業アプリケーションで広く採用されている。外部依存関係なしで多レベル設定とサードパーティライブラリとの統一管理が可能。
ライブラリ
logging (Standard Library)
概要
Python標準ライブラリの強力なロギングフレームワークです。設定可能性と拡張性に優れ、複雑な企業アプリケーションで広く採用されている。外部依存関係なしで多レベル設定とサードパーティライブラリとの統一管理が可能。2025年でも業界標準として安定した地位を維持し、複雑なシステムでの需要は継続的に高く、サードパーティライブラリとの互換性が重要視される環境では不可欠です。学習コストは高いが、習得すれば非常に強力なロギングソリューションを提供します。
詳細
Python logging 2025年版は、15年以上の成熟期間を経て、エンタープライズレベルのロギング要件を満たす包括的な機能セットを提供しています。ハンドラー、フォーマッター、フィルターの階層的な構成により、複雑なログルーティングと変換が可能。設定ファイル（YAML、JSON、INI）またはプログラマティック設定により、柔軟な運用環境対応を実現。マルチプロセス、マルチスレッド環境での安全性と、Django、Flask、FastAPI等主要フレームワークとのシームレス統合により、Pythonエコシステムの中核をなしています。
主な特徴
- 階層的ロガー構造: ドット記法による論理的なロガー階層管理
- 豊富なハンドラー: ファイル、コンソール、ネットワーク、メール等多様な出力先
- 柔軟なフォーマッター: カスタマイズ可能なログメッセージ形式
- フィルター機能: 条件ベースのログ処理制御
- 設定外部化: YAML、JSON、INIファイルによる設定管理
- マルチプロセス対応: 分散処理環境での安全なログ管理
メリット・デメリット
メリット
- 外部依存なしでPython環境に標準搭載
- 豊富な設定オプションによる高度なカスタマイズが可能
- エンタープライズ級アプリケーションでの豊富な実績
- サードパーティライブラリとの優れた統合性
- 詳細なドキュメントと豊富な学習リソース
- マルチプロセス・マルチスレッド環境での安全性保証
デメリット
- 設定の複雑さにより学習コストが高い
- 簡単な用途には機能が過剰で導入が煩雑
- デフォルト設定では現代的でないプレーンテキスト出力
- 構造化ログには追加設定や外部ライブラリが必要
- パフォーマンス重視の場面では他の選択肢が優位
- 初心者には概念理解が困難な場合がある
参考ページ
書き方の例
基本的なロギング設定
import logging
# 基本設定（最もシンプル）
logging.basicConfig(level=logging.DEBUG)
# より詳細な基本設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='app.log',
filemode='w'
)
# 基本的なログ出力
logging.debug('デバッグメッセージ')
logging.info('情報メッセージ')
logging.warning('警告メッセージ')
logging.error('エラーメッセージ')
logging.critical('重大エラーメッセージ')
# 例外情報付きログ
try:
result = 1 / 0
except ZeroDivisionError:
logging.exception('ゼロ除算エラーが発生しました')
# 変数を含むログ
user_id = 12345
action = 'login'
logging.info('ユーザー %s が %s を実行しました', user_id, action)
# f-string使用（Python 3.6以降）
logging.info(f'ユーザー {user_id} が {action} を実行しました')
階層的ロガーとカスタム設定
import logging
import logging.handlers
from datetime import datetime
# 階層的ロガーの作成
app_logger = logging.getLogger('myapp')
db_logger = logging.getLogger('myapp.database')
api_logger = logging.getLogger('myapp.api')
auth_logger = logging.getLogger('myapp.auth')
# カスタムフォーマッターの作成
class ColoredFormatter(logging.Formatter):
"""カラー出力対応フォーマッター"""
COLORS = {
'DEBUG': '\033[36m', # Cyan
'INFO': '\033[32m', # Green
'WARNING': '\033[33m', # Yellow
'ERROR': '\033[31m', # Red
'CRITICAL': '\033[35m' # Magenta
}
RESET = '\033[0m'
def format(self, record):
log_color = self.COLORS.get(record.levelname, self.RESET)
record.levelname = f"{log_color}{record.levelname}{self.RESET}"
return super().format(record)
# 複数ハンドラーの設定
def setup_logging():
# ルートロガーの設定
logging.getLogger().setLevel(logging.DEBUG)
# コンソールハンドラー（カラー出力）
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_formatter = ColoredFormatter(
'%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
console_handler.setFormatter(console_formatter)
# ファイルハンドラー（詳細ログ）
file_handler = logging.FileHandler('app_detailed.log', encoding='utf-8')
file_handler.setLevel(logging.DEBUG)
file_formatter = logging.Formatter(
'%(asctime)s [%(levelname)s] %(name)s - %(funcName)s:%(lineno)d - %(message)s'
)
file_handler.setFormatter(file_formatter)
# エラー専用ファイルハンドラー
error_handler = logging.FileHandler('errors.log', encoding='utf-8')
error_handler.setLevel(logging.ERROR)
error_formatter = logging.Formatter(
'%(asctime)s [%(levelname)s] %(name)s - %(pathname)s:%(lineno)d - %(message)s\n'
'Exception: %(exc_info)s\n'
)
error_handler.setFormatter(error_formatter)
# 回転ファイルハンドラー（サイズベース）
rotating_handler = logging.handlers.RotatingFileHandler(
'app_rotating.log',
maxBytes=1024*1024, # 1MB
backupCount=5,
encoding='utf-8'
)
rotating_handler.setLevel(logging.INFO)
rotating_formatter = logging.Formatter(
'%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
rotating_handler.setFormatter(rotating_formatter)
# 全ハンドラーをルートロガーに追加
root_logger = logging.getLogger()
root_logger.addHandler(console_handler)
root_logger.addHandler(file_handler)
root_logger.addHandler(error_handler)
root_logger.addHandler(rotating_handler)
# 設定適用とテスト
setup_logging()
# 階層的ロガーのテスト
app_logger.info('アプリケーション開始')
db_logger.debug('データベース接続初期化')
api_logger.info('API サーバー起動')
auth_logger.warning('認証トークンの有効期限が近づいています')
# エラーログのテスト
try:
raise ValueError('テスト例外')
except ValueError:
app_logger.exception('アプリケーションでエラーが発生')
設定ファイルによる管理
import logging
import logging.config
import yaml
import json
from pathlib import Path
# YAML設定ファイル (logging_config.yaml)
yaml_config = """
version: 1
disable_existing_loggers: false
formatters:
standard:
format: "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
detailed:
format: "%(asctime)s [%(levelname)s] %(name)s - %(funcName)s:%(lineno)d - %(message)s"
json:
format: "%(asctime)s"
class: pythonjsonlogger.jsonlogger.JsonFormatter
handlers:
console:
class: logging.StreamHandler
level: INFO
formatter: standard
stream: ext://sys.stdout
file:
class: logging.FileHandler
level: DEBUG
formatter: detailed
filename: app.log
mode: a
encoding: utf-8
rotating_file:
class: logging.handlers.RotatingFileHandler
level: INFO
formatter: standard
filename: app_rotating.log
maxBytes: 1048576 # 1MB
backupCount: 5
encoding: utf-8
time_rotating:
class: logging.handlers.TimedRotatingFileHandler
level: WARNING
formatter: detailed
filename: app_daily.log
when: midnight
interval: 1
backupCount: 30
encoding: utf-8
loggers:
myapp:
level: DEBUG
handlers: [console, file, rotating_file]
propagate: false
myapp.database:
level: DEBUG
handlers: [file]
propagate: true
myapp.api:
level: INFO
handlers: [console, rotating_file]
propagate: false
requests:
level: WARNING
handlers: [file]
propagate: false
root:
level: INFO
handlers: [console, time_rotating]
"""
# JSON設定ファイル (logging_config.json)
json_config = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"standard": {
"format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
},
"detailed": {
"format": "%(asctime)s [%(levelname)s] %(name)s - %(funcName)s:%(lineno)d - %(message)s"
}
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": "INFO",
"formatter": "standard",
"stream": "ext://sys.stdout"
},
"file": {
"class": "logging.FileHandler",
"level": "DEBUG",
"formatter": "detailed",
"filename": "app.log",
"mode": "a",
"encoding": "utf-8"
}
},
"loggers": {
"myapp": {
"level": "DEBUG",
"handlers": ["console", "file"],
"propagate": False
}
},
"root": {
"level": "INFO",
"handlers": ["console"]
}
}
class ConfigurableLoggingDemo:
def __init__(self):
self.logger = logging.getLogger('myapp')
def load_yaml_config(self):
"""YAML設定ファイルからロード"""
config = yaml.safe_load(yaml_config)
logging.config.dictConfig(config)
self.logger.info('YAML設定ファイルが読み込まれました')
def load_json_config(self):
"""JSON設定ファイルからロード"""
logging.config.dictConfig(json_config)
self.logger.info('JSON設定ファイルが読み込まれました')
def load_external_config(self, config_path: str):
"""外部設定ファイルからロード"""
path = Path(config_path)
if path.suffix.lower() == '.yaml' or path.suffix.lower() == '.yml':
with open(path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
logging.config.dictConfig(config)
elif path.suffix.lower() == '.json':
with open(path, 'r', encoding='utf-8') as f:
config = json.load(f)
logging.config.dictConfig(config)
elif path.suffix.lower() == '.ini' or path.suffix.lower() == '.cfg':
logging.config.fileConfig(path)
self.logger.info(f'設定ファイル {config_path} が読み込まれました')
def demo_logging(self):
"""ログ出力デモ"""
db_logger = logging.getLogger('myapp.database')
api_logger = logging.getLogger('myapp.api')
self.logger.debug('デバッグメッセージ')
self.logger.info('情報メッセージ')
self.logger.warning('警告メッセージ')
self.logger.error('エラーメッセージ')
db_logger.info('データベース操作完了')
api_logger.warning('API レスポンス時間が閾値を超えました')
# 使用例
demo = ConfigurableLoggingDemo()
print("=== YAML設定デモ ===")
demo.load_yaml_config()
demo.demo_logging()
print("\n=== JSON設定デモ ===")
demo.load_json_config()
demo.demo_logging()
高度なログ処理とフィルター
import logging
import time
from typing import Dict, Any
import threading
import queue
import json
class ContextFilter(logging.Filter):
"""コンテキスト情報を追加するフィルター"""
def __init__(self):
super().__init__()
self.context = threading.local()
def set_context(self, **kwargs):
"""コンテキスト情報を設定"""
for key, value in kwargs.items():
setattr(self.context, key, value)
def filter(self, record):
# コンテキスト情報をログレコードに追加
for key in dir(self.context):
if not key.startswith('_'):
setattr(record, key, getattr(self.context, key))
return True
class PerformanceFilter(logging.Filter):
"""パフォーマンス計測フィルター"""
def __init__(self, max_duration_ms: float = 1000.0):
super().__init__()
self.max_duration_ms = max_duration_ms
def filter(self, record):
# パフォーマンス情報がある場合のみ処理
if hasattr(record, 'duration_ms'):
# 閾値を超えた場合はWARNINGレベルに昇格
if record.duration_ms > self.max_duration_ms:
record.levelno = logging.WARNING
record.levelname = 'WARNING'
return True
class StructuredFormatter(logging.Formatter):
"""構造化ログフォーマッター"""
def format(self, record):
log_entry = {
'timestamp': self.formatTime(record),
'level': record.levelname,
'logger': record.name,
'message': record.getMessage(),
'module': record.module,
'function': record.funcName,
'line': record.lineno
}
# 例外情報の追加
if record.exc_info:
log_entry['exception'] = self.formatException(record.exc_info)
# カスタム属性の追加
for key, value in record.__dict__.items():
if key not in ['name', 'msg', 'args', 'levelname', 'levelno',
'pathname', 'filename', 'module', 'exc_info',
'exc_text', 'stack_info', 'lineno', 'funcName',
'created', 'msecs', 'relativeCreated', 'thread',
'threadName', 'processName', 'process', 'message']:
log_entry[key] = value
return json.dumps(log_entry, ensure_ascii=False, indent=None)
class AsyncLogHandler(logging.Handler):
"""非同期ログハンドラー"""
def __init__(self, handler):
super().__init__()
self.handler = handler
self.queue = queue.Queue()
self.thread = threading.Thread(target=self._worker)
self.thread.daemon = True
self.thread.start()
def emit(self, record):
"""ログレコードをキューに追加"""
self.queue.put(record)
def _worker(self):
"""バックグラウンドでログを処理"""
while True:
try:
record = self.queue.get()
if record is None:
break
self.handler.emit(record)
self.queue.task_done()
except Exception as e:
print(f"AsyncLogHandler error: {e}")
class AdvancedLoggingService:
def __init__(self):
self.context_filter = ContextFilter()
self.performance_filter = PerformanceFilter(max_duration_ms=500.0)
self.setup_logging()
def setup_logging(self):
"""高度なロギング設定"""
# ロガー取得
self.logger = logging.getLogger('advanced_app')
self.logger.setLevel(logging.DEBUG)
# 構造化ログハンドラー
structured_handler = logging.FileHandler('structured.log', encoding='utf-8')
structured_handler.setFormatter(StructuredFormatter())
structured_handler.addFilter(self.context_filter)
structured_handler.addFilter(self.performance_filter)
# 非同期ハンドラー
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter(
'%(asctime)s [%(levelname)s] %(name)s: %(message)s'
))
async_handler = AsyncLogHandler(console_handler)
async_handler.addFilter(self.context_filter)
# ハンドラー追加
self.logger.addHandler(structured_handler)
self.logger.addHandler(async_handler)
def process_request(self, request_id: str, user_id: int, action: str):
"""リクエスト処理（コンテキスト付きログ）"""
# コンテキスト設定
self.context_filter.set_context(
request_id=request_id,
user_id=user_id,
session_id=f"sess_{int(time.time())}"
)
start_time = time.time()
try:
self.logger.info(f'リクエスト処理開始: {action}')
# ビジネスロジック処理
self._execute_business_logic(action)
# パフォーマンス計測
duration_ms = (time.time() - start_time) * 1000
self.logger.info(
'リクエスト処理完了',
extra={'duration_ms': duration_ms, 'action': action}
)
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
self.logger.error(
f'リクエスト処理エラー: {action}',
exc_info=True,
extra={'duration_ms': duration_ms, 'action': action, 'error_type': type(e).__name__}
)
def _execute_business_logic(self, action: str):
"""ビジネスロジック実行"""
if action == 'slow_operation':
time.sleep(0.6) # 意図的に遅い処理
elif action == 'error_operation':
raise ValueError('意図的なエラー')
else:
time.sleep(0.1) # 通常処理
# 使用例とデモ
def advanced_logging_demo():
service = AdvancedLoggingService()
# 様々なケースのテスト
test_cases = [
('req_001', 12345, 'normal_operation'),
('req_002', 67890, 'slow_operation'),
('req_003', 11111, 'error_operation'),
('req_004', 22222, 'normal_operation'),
]
for request_id, user_id, action in test_cases:
service.process_request(request_id, user_id, action)
time.sleep(0.1) # 処理間隔
print("高度なロギングデモが完了しました")
time.sleep(1) # 非同期ログ処理の完了待ち
if __name__ == '__main__':
advanced_logging_demo()
Webアプリケーション統合例
import logging
import logging.config
from flask import Flask, request, g
import uuid
import time
from functools import wraps
# Flask アプリケーション用ロギング設定
LOGGING_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'default': {
'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
},
'detailed': {
'format': '%(asctime)s [%(levelname)s] %(name)s - %(funcName)s:%(lineno)d - %(message)s'
},
'access': {
'format': '%(asctime)s [%(levelname)s] %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'INFO',
'formatter': 'default'
},
'app_file': {
'class': 'logging.handlers.RotatingFileHandler',
'level': 'DEBUG',
'formatter': 'detailed',
'filename': 'app.log',
'maxBytes': 1048576,
'backupCount': 5,
'encoding': 'utf-8'
},
'access_file': {
'class': 'logging.handlers.TimedRotatingFileHandler',
'level': 'INFO',
'formatter': 'access',
'filename': 'access.log',
'when': 'midnight',
'interval': 1,
'backupCount': 30,
'encoding': 'utf-8'
}
},
'loggers': {
'app': {
'level': 'DEBUG',
'handlers': ['console', 'app_file'],
'propagate': False
},
'access': {
'level': 'INFO',
'handlers': ['access_file'],
'propagate': False
},
'werkzeug': {
'level': 'WARNING',
'handlers': ['console'],
'propagate': False
}
},
'root': {
'level': 'INFO',
'handlers': ['console']
}
}
# ロギング設定適用
logging.config.dictConfig(LOGGING_CONFIG)
app = Flask(__name__)
app_logger = logging.getLogger('app')
access_logger = logging.getLogger('access')
def log_requests(f):
"""リクエストログデコレーター"""
@wraps(f)
def decorated_function(*args, **kwargs):
# リクエストID生成
request_id = str(uuid.uuid4())
g.request_id = request_id
# リクエスト開始ログ
start_time = time.time()
app_logger.info(
f'[{request_id}] リクエスト開始: {request.method} {request.path}'
)
try:
# 実際の処理実行
response = f(*args, **kwargs)
# 成功ログ
duration = (time.time() - start_time) * 1000
access_logger.info(
f'{request.remote_addr} - "{request.method} {request.path}" '
f'200 {duration:.2f}ms [{request_id}]'
)
app_logger.info(f'[{request_id}] リクエスト完了: {duration:.2f}ms')
return response
except Exception as e:
# エラーログ
duration = (time.time() - start_time) * 1000
access_logger.error(
f'{request.remote_addr} - "{request.method} {request.path}" '
f'500 {duration:.2f}ms [{request_id}] Error: {str(e)}'
)
app_logger.exception(f'[{request_id}] リクエストエラー: {duration:.2f}ms')
raise
return decorated_function
@app.route('/')
@log_requests
def index():
app_logger.info(f'[{g.request_id}] インデックスページアクセス')
return 'Hello, World!'
@app.route('/users/<int:user_id>')
@log_requests
def get_user(user_id):
app_logger.info(f'[{g.request_id}] ユーザー情報取得: {user_id}')
if user_id == 999:
app_logger.error(f'[{g.request_id}] 不正なユーザーID: {user_id}')
raise ValueError('Invalid user ID')
return f'User: {user_id}'
@app.route('/slow')
@log_requests
def slow_endpoint():
app_logger.info(f'[{g.request_id}] 重い処理開始')
time.sleep(2) # 重い処理のシミュレーション
app_logger.info(f'[{g.request_id}] 重い処理完了')
return 'Slow operation completed'
@app.errorhandler(404)
def not_found(error):
request_id = getattr(g, 'request_id', 'unknown')
access_logger.warning(
f'{request.remote_addr} - "{request.method} {request.path}" '
f'404 [{request_id}]'
)
return 'Not Found', 404
@app.errorhandler(500)
def internal_error(error):
request_id = getattr(g, 'request_id', 'unknown')
app_logger.error(f'[{request_id}] 内部サーバーエラー', exc_info=True)
return 'Internal Server Error', 500
if __name__ == '__main__':
app_logger.info('Flaskアプリケーション開始')
app.run(debug=True, host='0.0.0.0', port=5000)