Flaskでの静的ファイルの取り扱い
はじめに FlaskはPythonでWebアプリケーションを開発するための軽量で柔軟なフレームワークです。Webサイトを作成する際には、HTMLだけでなくCSS […]
Flaskにおけるエラーハンドリングとデバッグ
2026-03-05はじめに
FlaskはPythonの軽量なWebフレームワークとして、シンプルさと柔軟性から多くの開発者に支持されています。しかし、実際の運用環境では予期せぬエラーや例外が発生するものです。適切なエラーハンドリングと効果的なデバッグ技法を実装することは、ユーザー体験の向上と開発効率の改善に不可欠です。本記事では、Flaskアプリケーションにおけるエラーハンドリングの実装方法と、開発・本番環境での効果的なデバッグ手法について詳細に解説します。
カスタムエラーページの実装
Flaskでは、デフォルトで標準的なエラーページが提供されていますが、これらはユーザーフレンドリーとは言えません。カスタムエラーページを実装することで、ユーザーに適切なフィードバックを提供し、ブランドイメージを維持することができます。
基本的なエラーハンドラの実装
Flaskでは@app.errorhandlerデコレータを使用して、特定のHTTPステータスコードに対するカスタムハンドラを定義できます。
from flask import Flask, render_template
app = Flask(__name__)
@app.errorhandler(404)
def page_not_found(error):
return render_template('404.html'), 404
@app.errorhandler(500)
def internal_server_error(error):
return render_template('500.html'), 500
@app.errorhandler(403)
def forbidden(error):
return render_template('403.html'), 403テンプレートの作成
エラーページ用のテンプレートは、通常のテンプレートと同様に作成しますが、ユーザーが次に取るべき行動を示すことが重要です。
<!-- templates/404.html -->
<!DOCTYPE html>
<html>
<head>
<title>ページが見つかりません</title>
</head>
<body>
<div class="error-container">
<h1>404 - ページが見つかりません</h1>
<p>お探しのページは存在しないか、移動した可能性があります。</p>
<a href="{{ url_for('index') }}">ホームページに戻る</a>
</div>
</body>
</html>アプリケーションエラーのハンドリング
HTTPエラーだけでなく、アプリケーション固有の例外もハンドリングすることができます。
class ValidationError(Exception):
pass
@app.errorhandler(ValidationError)
def handle_validation_error(error):
response = jsonify({'error': 'Validation failed', 'message': str(error)})
response.status_code = 400
return responseグローバルな例外ハンドリング
@app.errorhandler(Exception)を使用して、すべての未処理例外をキャッチすることも可能ですが、慎重に使用する必要があります。
@app.errorhandler(Exception)
def handle_unexpected_error(error):
app.logger.error(f'Unhandled exception: {error}')
return render_template('500.html'), 500ロギング設定
効果的なロギングは、アプリケーションの状態を監視し、問題を迅速に特定するための重要な手段です。
Flaskのデフォルトロガーの設定
Flaskは標準ライブラリのloggingモジュールを使用しています。アプリケーション起動時にロギング設定を初期化できます。
import logging
from logging.handlers import RotatingFileHandler
def setup_logging(app):
# 基本的な設定
logging.basicConfig(level=logging.INFO)
# ファイルハンドラの設定（ログローテーション付き）
file_handler = RotatingFileHandler(
'app.log',
maxBytes=1024 * 1024 * 10, # 10MB
backupCount=10
)
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
# アプリケーションロガーにハンドラを追加
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
# リクエストごとのロギング
@app.before_request
def log_request_info():
app.logger.info(f'Request: {request.method} {request.path}')
@app.after_request
def log_response_info(response):
app.logger.info(f'Response: {response.status}')
return response
# アプリケーション初期化時にロギングをセットアップ
setup_logging(app)構造化ロギングの実装
JSON形式のログは、ログ管理システムとの連携に適しています。
import json
import logging
class JsonFormatter(logging.Formatter):
def format(self, record):
log_record = {
'timestamp': self.formatTime(record),
'level': record.levelname,
'message': record.getMessage(),
'module': record.module,
'funcName': record.funcName,
'lineno': record.lineno,
}
# 追加のコンテキストがある場合は追加
if hasattr(record, 'extra_data'):
log_record.update(record.extra_data)
return json.dumps(log_record)
# フォーマッタの適用
json_formatter = JsonFormatter()
file_handler.setFormatter(json_formatter)環境別のロギングレベル
開発環境と本番環境で異なるロギングレベルを設定します。
import os
def configure_logging_based_on_env(app):
env = os.getenv('FLASK_ENV', 'production')
if env == 'development':
app.logger.setLevel(logging.DEBUG)
# コンソールにも出力
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
app.logger.addHandler(console_handler)
else:
app.logger.setLevel(logging.WARNING)デバッグ技法
効率的なデバッグは、開発速度を大幅に向上させます。
Flaskデバッグモードの活用
開発時にはデバッグモードを有効にします。
if __name__ == '__main__':
app.run(debug=True)または環境変数で設定：
export FLASK_ENV=development
export FLASK_DEBUG=1Werkzeugデバッガの使用
FlaskのデフォルトデバッガであるWerkzeugは、ブラウザ上でインタラクティブなデバッグを可能にします。エラー発生時に表示されるトレースバックページでは、各スタックフレームのコードを表示し、Pythonシェルを開いて変数を検査できます。
デバッグツールバーの導入
Flask-DebugToolbarは、リクエスト/レスポンス情報、SQLクエリ、テンプレート情報などを可視化する強力なツールです。
from flask_debugtoolbar import DebugToolbarExtension
app.config['SECRET_KEY'] = 'your-secret-key'
app.config['DEBUG_TB_INTERCEPT_REDIRECTS'] = False
toolbar = DebugToolbarExtension(app)カスタムデバッグ関数の作成
特定の状況でデバッグ情報を出力するためのユーティリティ関数を作成します。
import pprint
from flask import current_app
def debug_data(label, data):
"""開発環境でのみデータをログ出力する"""
if current_app.config.get('DEBUG'):
current_app.logger.debug(f"=== {label} ===")
current_app.logger.debug(pprint.pformat(data))
current_app.logger.debug("=" * (len(label) + 8))
# 使用例
@app.route('/debug-example')
def debug_example():
user_data = {'id': 1, 'name': 'Test User', 'roles': ['admin', 'user']}
debug_data('User Data', user_data)
return 'Check logs for debug info'リクエスト/レスポンスのインターセプト
デバッグのために、リクエスト前後の処理をフックすることができます。
@app.before_request
def before_request_debug():
if app.debug:
print(f"Request: {request.method} {request.url}")
print(f"Headers: {dict(request.headers)}")
print(f"Body: {request.get_data()}")
@app.after_request
def after_request_debug(response):
if app.debug:
print(f"Response: {response.status}")
print(f"Response Headers: {dict(response.headers)}")
return responseパフォーマンスプロファイリング
応答時間のボトルネックを特定するためのプロファイリング。
import time
from functools import wraps
def profile(endpoint):
@wraps(endpoint)
def wrapper(*args, **kwargs):
start_time = time.time()
result = endpoint(*args, **kwargs)
elapsed_time = time.time() - start_time
current_app.logger.info(
f"Endpoint {endpoint.__name__} took {elapsed_time:.4f} seconds"
)
return result
return wrapper
# 使用例
@app.route('/profile-me')
@profile
def profiled_endpoint():
# 何らかの処理
time.sleep(0.5)
return 'Profiled response'テスト環境でのデバッグ
テスト時に特定の状態を再現するためのデバッグ機能。
@app.route('/_debug/set-session//')
def debug_set_session(key, value):
"""開発環境でのみセッションを設定できるエンドポイント"""
if not app.debug:
return 'Not available in production', 403
session[key] = value
return f'Session set: {key} = {value}'
@app.route('/_debug/raise-error/')
def debug_raise_error(error_type):
"""開発環境でのみエラーを発生させるテストエンドポイント"""
if not app.debug:
return 'Not available in production', 403
if error_type == '500':
raise Exception('Test 500 error')
elif error_type == '404':
abort(404)
return 'Valid error types: 500, 404' まとめ
Flaskアプリケーションにおける効果的なエラーハンドリングとデバッグは、アプリケーションの品質と開発効率を大幅に向上させます。カスタムエラーページの実装により、ユーザーはエラー発生時にも適切なガイダンスを受けられ、ブランド体験を維持できます。適切なロギング設定は、本番環境での問題追跡と分析を可能にし、迅速な問題解決を支援します。さまざまなデバッグ技法を組み合わせることで、開発サイクルを効率化し、より堅牢なアプリケーションを構築することができます。
これらのテクニックは、小規模なプロトタイプから大規模な本番アプリケーションまで、あらゆる規模のFlaskプロジェクトに適用可能です。開発初期段階からこれらのプラクティスを導入することで、技術的負債の蓄積を防ぎ、長期的なメンテナンスコストを削減することができるでしょう。エラーハンドリングとデバッグは単なる「問題対応」ではなく、品質の高いソフトウェア開発の基盤であることを常に意識して実装を進めることが重要です。
演習問題
初級問題（3問）
初級1: カスタム404エラーページの実装
次の要件を満たすFlaskアプリケーションを作成してください。
/にアクセスすると「ようこそ」と表示される- 存在しないパスにアクセスした場合、カスタムの404エラーページを表示する
- 404エラーページには「ページが見つかりません」というメッセージと、トップページへのリンクを含める
- 404エラーページは
404.htmlテンプレートを使用すること
初級2: 基本的なロギング設定
以下の要件を満たすロギング設定を実装してください。
- アプリケーション起動時に「アプリケーションが起動しました」というINFOレベルのログを出力する
- すべてのリクエストについて「[メソッド] [パス] にアクセスがありました」というDEBUGレベルのログを出力する
- ログはコンソールと
app.logファイルの両方に出力する - ファイルログは10MBでローテーションし、最大10個のバックアップを保持する
初級3: デバッグモードの理解
次の質問に答えてください。
- Flaskのデバッグモードを有効にする2つの方法を説明してください。
- デバッグモードを有効にした場合、アプリケーションにどのような変化が起こりますか？
- 本番環境でデバッグモードを有効にすることの危険性は何ですか？
中級問題（6問）
中級1: 複数のエラーページ実装
次のHTTPエラーコードに対するカスタムエラーページを実装してください。
- 403（Forbidden）: アクセス権限がないことを伝えるページ
- 500（Internal Server Error）: サーバーエラーが発生したことを伝えるページ
- 400（Bad Request）: リクエストが不正であることを伝えるページ
各エラーページには適切なメッセージとナビゲーションを設け、対応するテンプレートファイルを使用してください。
中級2: アプリケーション例外のハンドリング
次の要件を満たすカスタム例外ハンドリングを実装してください。
ValidationErrorというカスタム例外クラスを作成する- この例外が発生した場合、JSON形式で
{"error": "validation_failed", "message": [エラーメッセージ]}を返す - レスポンスのステータスコードは400とする
/validateエンドポイントを作成し、クエリパラメータnumberが整数でない場合にValidationErrorを発生させる
中級3: 環境別ロギング設定
以下の要件を満たす環境別ロギング設定を実装してください。
- 開発環境（FLASK_ENV=development）ではDEBUGレベル以上のログをコンソールに出力する
- 本番環境（FLASK_ENV=production）ではWARNINGレベル以上のログをファイルに出力する
- 開発環境では各リクエストの詳細（メソッド、パス、クエリパラメータ）をログ出力する
- 環境変数
LOG_LEVELが設定されている場合は、そのレベルを優先する
中級4: 構造化ロギングの実装
JSON形式で構造化ログを出力するロギングシステムを実装してください。
- 各ログエントリには
timestamp,level,message,endpoint,user_id（認証されている場合）を含める - ログフォーマッタは
JsonFormatterクラスとして実装する - エラーログには追加で
traceback情報を含める - リクエストIDを生成し、同じリクエストに関連するすべてのログに含める
中級5: デバッグ用エンドポイントの作成
開発環境のみで利用できるデバッグ用エンドポイントを実装してください。
/_debug/sessionで現在のセッション内容を表示する/_debug/configでアプリケーション設定を表示する（パスワードなどの機密情報はマスクする）/_debug/routesで登録されているすべてのルートを表示する- これらのエンドポイントは本番環境では403エラーを返す
中級6: パフォーマンス計測デコレータ
以下の機能を持つパフォーマンス計測デコレータを作成してください。
- デコレートされたエンドポイントの実行時間を計測する
- 実行時間が閾値（デフォルト1秒）を超えた場合、WARNINGログを出力する
- 計測結果を
response.headers['X-Execution-Time']に追加する - デコレータは任意の閾値設定を可能にする
上級問題（3問）
上級1: エラーハンドリングミドルウェアの作成
次の要件を満たすエラーハンドリングミドルウェアクラスを作成してください。
ErrorHandlingMiddlewareクラスを実装し、WSGIミドルウェアとして機能させる- すべての未処理例外をキャッチし、適切なエラーレスポンスに変換する
- エラーログにはリクエスト情報（IP、ユーザーエージェント、パスなど）を含める
- エラーの種類に応じて異なるハンドリング（APIリクエスト vs HTMLリクエスト）を行う
- カスタムエラーページのレンダリング時にテンプレートコンテキストを自動で渡す
上級2: 分散トレーシング対応ロギングシステム
マイクロサービス環境を想定した分散トレーシング対応のロギングシステムを実装してください。
- リクエストIDを生成し、すべての関連ログに自動で追加する
- 外部サービス呼び出しのログを記録し、呼び出し時間と結果を追跡する
- ログエントリに関連サービスや親リクエストIDを含める
- ログ集計用にPrometheus形式のメトリクスも出力する
- 非同期タスクのログも同じトレースIDで追跡できるようにする
上級3: リアルタイムデバッグ・モニタリングダッシュボード
Flaskアプリケーションのリアルタイムデバッグ・モニタリングダッシュボードを実装してください。
- WebSocketを使用したリアルタイムログ表示機能
- 現在のアプリケーション状態（メモリ使用量、CPU使用率、アクティブセッション数など）の可視化
- リクエスト/レスポンスのライブトレース機能
- SQLクエリの実行計画とパフォーマンス分析
- 動的なログレベル変更機能（実行中のアプリケーションでログレベルを変更可能）
解答のヒント
初級問題
@app.errorhandler(404)デコレータとrender_template()関数を使用logging.basicConfig()とRotatingFileHandlerの組み合わせ- 公式ドキュメントの「デバッグモード」セクションを参照
中級問題
@app.errorhandler()を各ステータスコードに対して実装- カスタム例外クラスと
@app.errorhandler()の組み合わせ os.getenv()で環境変数を取得し、条件分岐logging.Formatterクラスを継承してカスタムフォーマッタを作成app.debugフラグをチェックし、開発環境のみ機能を有効化time.time()とfunctools.wrapsを使用
上級問題
__call__メソッドを実装したWSGIミドルウェアクラスを作成- コンテキストマネージャとスレッドローカルストレージを使用
- Flask-SocketIOやFlask-Adminなどの拡張を参考に、独自の管理画面を構築
初級問題 解答例
初級1: カスタム404エラーページの実装
app.py
from flask import Flask, render_template
app = Flask(__name__)
@app.route('/')
def index():
return 'ようこそ'
@app.errorhandler(404)
def page_not_found(error):
return render_template('404.html'), 404
if __name__ == '__main__':
app.run(debug=True)templates/404.html
<!DOCTYPE html>
<html>
<head>
<title>ページが見つかりません</title>
</head>
<body>
<h1>404 - ページが見つかりません</h1>
<p>お探しのページは存在しないか、移動した可能性があります。</p>
<a href="/">トップページに戻る</a>
</body>
</html>初級2: 基本的なロギング設定
app.py
import logging
from logging.handlers import RotatingFileHandler
from flask import Flask, request
app = Flask(__name__)
def setup_logging():
# ロガーの基本設定
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# ファイルハンドラの設定
file_handler = RotatingFileHandler(
'app.log',
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=10,
encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
# コンソールハンドラの設定
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
# アプリケーションロガーにハンドラを追加
app.logger.addHandler(file_handler)
app.logger.addHandler(console_handler)
app.logger.setLevel(logging.DEBUG)
# アプリケーション起動ログ
app.logger.info('アプリケーションが起動しました')
@app.before_request
def log_request():
app.logger.debug(f'{request.method} {request.path} にアクセスがありました')
@app.route('/')
def index():
return 'ロギングデモ'
if __name__ == '__main__':
setup_logging()
app.run(debug=True)初級3: デバッグモードの理解
解答例
"""
1. Flaskのデバッグモードを有効にする2つの方法:
方法1: app.run()で直接指定
app.run(debug=True)
方法2: 環境変数で設定
export FLASK_ENV=development
export FLASK_DEBUG=1
または
import os
os.environ['FLASK_ENV'] = 'development'
os.environ['FLASK_DEBUG'] = '1'
2. デバッグモードを有効にした場合の変化:
- コード変更時の自動リロード
- 詳細なエラーページの表示（Werkzeugデバッガ）
- エラーページでのインタラクティブなデバッグシェル
- パフォーマンスの若干の低下（開発機能のため）
3. 本番環境でのデバッグモード有効化の危険性:
- スタックトレースや内部情報が公開される（情報漏洩）
- 任意のコード実行が可能になる（セキュリティリスク）
- アプリケーションのパフォーマンス低下
- デバッグシェルを通じたシステムアクセスが可能になる
"""中級問題 解答例
中級1: 複数のエラーページ実装
app.py
from flask import Flask, render_template, abort
app = Flask(__name__)
@app.route('/')
def index():
return 'ホームページ'
@app.route('/forbidden')
def trigger_403():
abort(403)
@app.route('/error')
def trigger_500():
raise Exception('テスト用エラー')
@app.route('/bad')
def trigger_400():
abort(400)
# エラーハンドラ
@app.errorhandler(403)
def forbidden_error(error):
return render_template('403.html'), 403
@app.errorhandler(500)
def internal_error(error):
return render_template('500.html'), 500
@app.errorhandler(400)
def bad_request_error(error):
return render_template('400.html'), 400
if __name__ == '__main__':
app.run(debug=True)templates/403.html
<!DOCTYPE html>
<html>
<head>
<title>アクセス拒否</title>
</head>
<body>
<h1>403 - アクセスが拒否されました</h1>
<p>このページにアクセスする権限がありません。</p>
<a href="/">ホームに戻る</a>
</body>
</html>templates/500.html
<!DOCTYPE html>
<html>
<head>
<title>サーバーエラー</title>
</head>
<body>
<h1>500 - 内部サーバーエラー</h1>
<p>サーバーでエラーが発生しました。しばらくしてから再度お試しください。</p>
<a href="/">ホームに戻る</a>
</body>
</html>templates/400.html
<!DOCTYPE html>
<html>
<head>
<title>不正なリクエスト</title>
</head>
<body>
<h1>400 - 不正なリクエスト</h1>
<p>リクエストが不正です。入力内容を確認してください。</p>
<a href="/">ホームに戻る</a>
</body>
</html>中級2: アプリケーション例外のハンドリング
app.py
from flask import Flask, request, jsonify
app = Flask(__name__)
# カスタム例外クラス
class ValidationError(Exception):
pass
# エラーハンドラ
@app.errorhandler(ValidationError)
def handle_validation_error(error):
response = jsonify({
'error': 'validation_failed',
'message': str(error)
})
response.status_code = 400
return response
@app.route('/validate')
def validate_number():
number_str = request.args.get('number', '')
try:
number = int(number_str)
return jsonify({
'status': 'success',
'number': number
})
except ValueError:
raise ValidationError(f'"{number_str}" は有効な整数ではありません')
@app.route('/')
def index():
return 'バリデーションAPI'
if __name__ == '__main__':
app.run(debug=True)中級3: 環境別ロギング設定
app.py
import os
import logging
from logging.handlers import RotatingFileHandler
from flask import Flask, request
app = Flask(__name__)
def configure_logging():
# 環境変数から設定を取得
env = os.getenv('FLASK_ENV', 'production')
log_level_str = os.getenv('LOG_LEVEL', '').upper()
# ログレベルの決定
log_levels = {
'DEBUG': logging.DEBUG,
'INFO': logging.INFO,
'WARNING': logging.WARNING,
'ERROR': logging.ERROR,
'CRITICAL': logging.CRITICAL
}
if log_level_str in log_levels:
log_level = log_levels[log_level_str]
elif env == 'development':
log_level = logging.DEBUG
else:
log_level = logging.WARNING
# ロガーの設定
app.logger.setLevel(log_level)
# フォーマッタ
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# 開発環境用のコンソールハンドラ
if env == 'development':
console_handler = logging.StreamHandler()
console_handler.setLevel(log_level)
console_handler.setFormatter(formatter)
app.logger.addHandler(console_handler)
# 本番環境用のファイルハンドラ
if env == 'production':
file_handler = RotatingFileHandler(
'production.log',
maxBytes=10 * 1024 * 1024,
backupCount=5,
encoding='utf-8'
)
file_handler.setLevel(log_level)
file_handler.setFormatter(formatter)
app.logger.addHandler(file_handler)
# リクエストロギング（開発環境のみ詳細）
@app.before_request
def log_request_details():
if env == 'development':
app.logger.debug(
f'リクエスト: {request.method} {request.path} '
f'クエリ: {dict(request.args)}'
)
else:
app.logger.info(f'リクエスト: {request.method} {request.path}')
@app.route('/')
def index():
app.logger.info('インデックスページにアクセス')
return '環境別ロギングデモ'
@app.route('/error')
def trigger_error():
app.logger.error('エラーが発生しました')
return 'エラーページ'
if __name__ == '__main__':
configure_logging()
app.run()中級4: 構造化ロギングの実装
app.py
import json
import logging
import traceback
import uuid
from flask import Flask, request, g
from functools import wraps
app = Flask(__name__)
class JsonFormatter(logging.Formatter):
def format(self, record):
log_data = {
'timestamp': self.formatTime(record),
'level': record.levelname,
'message': record.getMessage(),
'module': record.module,
'funcName': record.funcName,
'lineno': record.lineno,
}
# リクエスト情報の追加
if hasattr(g, 'request_id'):
log_data['request_id'] = g.request_id
# エンドポイント情報
if request:
log_data['endpoint'] = request.endpoint or 'unknown'
# ユーザー情報（認証されている場合）
if hasattr(g, 'user_id'):
log_data['user_id'] = g.user_id
# エラー時のトレースバック
if record.exc_info:
log_data['traceback'] = traceback.format_exception(*record.exc_info)
return json.dumps(log_data, ensure_ascii=False)
def setup_structured_logging():
# JSONフォーマッタの作成
formatter = JsonFormatter()
# コンソールハンドラ
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
# ファイルハンドラ
file_handler = logging.FileHandler('structured.log', encoding='utf-8')
file_handler.setFormatter(formatter)
# ロガーの設定
app.logger.addHandler(console_handler)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
@app.before_request
def before_request():
# リクエストIDの生成
g.request_id = str(uuid.uuid4())
# ダミーユーザーID（実際は認証から取得）
g.user_id = request.headers.get('X-User-ID', 'anonymous')
app.logger.info(f'リクエスト開始: {request.method} {request.path}')
@app.after_request
def after_request(response):
app.logger.info(f'リクエスト完了: {response.status_code}')
response.headers['X-Request-ID'] = g.request_id
return response
@app.route('/')
def index():
app.logger.info('インデックスページへアクセス')
return '構造化ロギングデモ'
@app.route('/error')
def trigger_error():
try:
raise ValueError('テストエラー')
except ValueError as e:
app.logger.error('エラーが発生しました', exc_info=True)
return 'エラーが発生しました', 500
if __name__ == '__main__':
setup_structured_logging()
app.run(debug=True)中級5: デバッグ用エンドポイントの作成
app.py
import re
from flask import Flask, jsonify, session, current_app
app = Flask(__name__)
app.config['SECRET_KEY'] = 'debug-secret-key'
app.config['DEBUG'] = True # 開発環境ではTrue
def development_only(f):
"""開発環境のみ実行を許可するデコレータ"""
from functools import wraps
@wraps(f)
def wrapper(*args, **kwargs):
if not current_app.debug:
return 'このエンドポイントは開発環境でのみ利用できます', 403
return f(*args, **kwargs)
return wrapper
@app.route('/')
def index():
session['test_key'] = 'test_value'
return 'デバッグエンドポイントデモ'
@app.route('/_debug/session')
@development_only
def debug_session():
"""セッション内容を表示"""
return jsonify(dict(session))
@app.route('/_debug/config')
@development_only
def debug_config():
"""設定を表示（機密情報はマスク）"""
config = {}
# 機密情報をマスクする正規表現
secret_patterns = [
r'.*key.*',
r'.*secret.*',
r'.*password.*',
r'.*token.*',
r'.*auth.*'
]
for key, value in current_app.config.items():
# 機密情報かチェック
is_secret = any(re.match(pattern, key.lower())
for pattern in secret_patterns)
if is_secret and value:
config[key] = '******** (機密情報のためマスクされています)'
else:
config[key] = str(value)
return jsonify(config)
@app.route('/_debug/routes')
@development_only
def debug_routes():
"""登録されているルートを表示"""
routes = []
for rule in current_app.url_map.iter_rules():
routes.append({
'endpoint': rule.endpoint,
'methods': list(rule.methods),
'rule': str(rule),
'arguments': list(rule.arguments)
})
return jsonify(routes)
@app.route('/_debug/test-error')
@development_only
def test_error():
"""デバッグ用エラー発生エンドポイント"""
raise Exception('デバッグ用テストエラー')
if __name__ == '__main__':
app.run(debug=True)中級6: パフォーマンス計測デコレータ
app.py
import time
import logging
from functools import wraps
from flask import Flask, request, g, jsonify
app = Flask(__name__)
def measure_performance(threshold=1.0):
"""
パフォーマンス計測デコレータ
Args:
threshold: 警告を出す実行時間の閾値（秒）
"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
# 計測開始
start_time = time.time()
try:
# 関数実行
result = f(*args, **kwargs)
return result
finally:
# 計測終了とログ出力
elapsed_time = time.time() - start_time
# 実行時間をgオブジェクトに保存（レスポンスヘッダー用）
g.execution_time = elapsed_time
# 閾値を超えた場合の警告ログ
if elapsed_time > threshold:
app.logger.warning(
f'パフォーマンス警告: {request.endpoint} '
f'実行時間 {elapsed_time:.3f}秒 > 閾値 {threshold}秒'
)
# 常に情報ログを出力
app.logger.info(
f'エンドポイント {request.endpoint} '
f'実行時間: {elapsed_time:.3f}秒'
)
return decorated_function
return decorator
@app.after_request
def add_performance_header(response):
"""レスポンスヘッダーに実行時間を追加"""
if hasattr(g, 'execution_time'):
response.headers['X-Execution-Time'] = f'{g.execution_time:.3f}s'
return response
@app.route('/fast')
@measure_performance(threshold=0.5) # 0.5秒以上の実行で警告
def fast_endpoint():
"""高速なエンドポイント"""
time.sleep(0.1) # 100ms待機
return jsonify({'status': 'fast', 'message': '高速処理完了'})
@app.route('/slow')
@measure_performance(threshold=1.0) # 1秒以上の実行で警告
def slow_endpoint():
"""低速なエンドポイント"""
time.sleep(1.5) # 1.5秒待機（閾値超え）
return jsonify({'status': 'slow', 'message': '低速処理完了'})
@app.route('/custom-threshold')
@measure_performance(threshold=2.0) # カスタム閾値
def custom_threshold_endpoint():
"""カスタム閾値のエンドポイント"""
time.sleep(1.8) # 1.8秒待機
return jsonify({'status': 'custom', 'message': 'カスタム処理完了'})
@app.route('/no-decorator')
def no_decorator_endpoint():
"""デコレータなしのエンドポイント（比較用）"""
return jsonify({'status': 'no_decorator'})
if __name__ == '__main__':
app.run(debug=True)上級問題 解答例
上級1: エラーハンドリングミドルウェアの作成
app.py
import traceback
import json
from flask import Flask, request, render_template, jsonify
from werkzeug.exceptions import HTTPException
class ErrorHandlingMiddleware:
"""
エラーハンドリングミドルウェア
すべての未処理例外をキャッチし、適切なエラーレスポンスに変換する
"""
def __init__(self, app):
self.app = app
self.app.wsgi_app = self
# エラーテンプレートのデフォルトコンテキスト
self.default_context = {
'app_name': 'Flask Application',
'support_email': 'support@example.com'
}
def __call__(self, environ, start_response):
try:
return self.app.wsgi_app(environ, start_response)
except Exception as e:
# リクエストコンテキストの設定
with self.app.request_context(environ):
return self.handle_exception(e, environ, start_response)
def handle_exception(self, exception, environ, start_response):
"""例外をハンドリングして適切なレスポンスを生成"""
# リクエスト情報の収集
request_info = self._collect_request_info(environ)
# ログ出力
self._log_error(exception, request_info)
# エラーレスポンスの生成
response = self._create_error_response(exception, request_info)
# WSGIレスポンスの生成
status = f"{response.status_code} {response.status}"
response_headers = [(k, v) for k, v in response.headers.items()]
start_response(status, response_headers)
return [response.get_data()]
def _collect_request_info(self, environ):
"""リクエスト情報を収集"""
return {
'method': environ.get('REQUEST_METHOD', ''),
'path': environ.get('PATH_INFO', ''),
'query_string': environ.get('QUERY_STRING', ''),
'remote_addr': environ.get('REMOTE_ADDR', ''),
'user_agent': environ.get('HTTP_USER_AGENT', ''),
'content_type': environ.get('CONTENT_TYPE', ''),
'accept': environ.get('HTTP_ACCEPT', ''),
'referer': environ.get('HTTP_REFERER', ''),
}
def _log_error(self, exception, request_info):
"""エラーログを出力"""
error_message = f"未処理例外: {type(exception).__name__}: {str(exception)}"
# リクエスト情報を含めたログ
log_message = f"{error_message}\nリクエスト情報: {request_info}\n"
# トレースバック
log_message += "トレースバック:\n" + traceback.format_exc()
self.app.logger.error(log_message)
def _create_error_response(self, exception, request_info):
"""エラータイプに応じたレスポンスを生成"""
# HTTP例外の処理
if isinstance(exception, HTTPException):
status_code = exception.code
message = exception.description
else:
status_code = 500
message = "Internal Server Error"
# リクエストのAcceptヘッダーをチェック
accept = request_info.get('accept', '').lower()
# JSONリクエストかどうかを判定
is_json_request = ('application/json' in accept or
request_info.get('content_type', '').lower() == 'application/json')
if is_json_request:
# JSONレスポンスを返す
response_data = {
'error': {
'code': status_code,
'message': message,
'type': type(exception).__name__
},
'request_id': request_info.get('request_id', ''),
'timestamp': datetime.now().isoformat()
}
# 開発環境の場合はスタックトレースを含める
if self.app.debug:
response_data['error']['traceback'] = traceback.format_exc().split('\n')
response = jsonify(response_data)
response.status_code = status_code
else:
# HTMLレスポンスを返す
context = self.default_context.copy()
context.update({
'error_code': status_code,
'error_message': message,
'request_info': request_info,
'is_debug': self.app.debug
})
# 開発環境の場合はトレースバックを追加
if self.app.debug:
context['traceback'] = traceback.format_exc()
# エラーテンプレートのレンダリング
template_name = f"errors/{status_code}.html"
try:
html = render_template(template_name, **context)
except:
# エラーテンプレートがない場合はデフォルトテンプレート
html = render_template('errors/default.html', **context)
response = self.app.response_class(
html,
status=status_code,
mimetype='text/html'
)
return response
# Flaskアプリケーションの作成
app = Flask(__name__)
app.wsgi_app = ErrorHandlingMiddleware(app)
# エラーテンプレート用のコンテキストプロセッサ
@app.context_processor
def inject_default_context():
return ErrorHandlingMiddleware.default_context
@app.route('/')
def index():
return 'エラーハンドリングミドルウェアデモ'
@app.route('/test-http-error/')
def test_http_error(code):
from werkzeug.exceptions import default_exceptions
if code in default_exceptions:
raise default_exceptions[code]()
return f"HTTP {code} エラーのテスト"
@app.route('/test-exception')
def test_exception():
raise ValueError("テスト用の例外です")
if __name__ == '__main__':
app.run(debug=True) templates/errors/default.html
<!DOCTYPE html>
<html>
<head>
<title>エラー {{ error_code }} - {{ app_name }}</title>
</head>
<body>
<h1>エラー {{ error_code }}: {{ error_message }}</h1>
{% if is_debug and traceback %}
<details>
<summary>詳細情報（開発環境のみ）</summary>
<pre>{{ traceback }}</pre>
</details>
{% endif %}
<p>問題が解決しない場合は、{{ support_email }}までご連絡ください。</p>
<a href="/">ホームに戻る</a>
</body>
</html>上級2: 分散トレーシング対応ロギングシステム
app.py
import uuid
import time
import json
import logging
import threading
from datetime import datetime
from contextvars import ContextVar
from functools import wraps
from flask import Flask, request, g, jsonify
import requests
# コンテキスト変数の定義
request_id_var = ContextVar('request_id', default=None)
parent_request_id_var = ContextVar('parent_request_id', default=None)
class DistributedLoggingSystem:
"""分散トレーシング対応ロギングシステム"""
def __init__(self, app=None):
self.app = app
self.local = threading.local()
if app is not None:
self.init_app(app)
def init_app(self, app):
self.app = app
# JSONフォーマッタの設定
self._setup_formatter()
# リクエスト前後のフックを設定
self._setup_hooks()
# Prometheusメトリクスの初期化
self._init_metrics()
def _setup_formatter(self):
"""JSONフォーマッタを設定"""
class TracingJsonFormatter(logging.Formatter):
def format(self, record):
log_data = {
'timestamp': datetime.utcnow().isoformat() + 'Z',
'level': record.levelname,
'message': record.getMessage(),
'service': 'flask-app',
'trace': {}
}
# トレース情報の追加
trace_info = self._get_trace_info()
log_data['trace'].update(trace_info)
# 追加フィールド
if hasattr(record, 'extra_fields'):
log_data.update(record.extra_fields)
# エラー情報
if record.exc_info:
log_data['error'] = {
'type': record.exc_info[0].__name__,
'message': str(record.exc_info[1]),
'stack_trace': self.formatException(record.exc_info)
}
return json.dumps(log_data, ensure_ascii=False)
def _get_trace_info(self):
"""トレース情報を取得"""
trace_info = {
'request_id': request_id_var.get(),
'parent_request_id': parent_request_id_var.get(),
'span_id': str(uuid.uuid4())[:8]
}
# 現在のスレッドから追加情報を取得
if hasattr(threading.local(), 'service_context'):
trace_info.update(threading.local().service_context)
return trace_info
# フォーマッタの適用
formatter = TracingJsonFormatter()
# ハンドラの設定
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
file_handler = logging.FileHandler('distributed.log', encoding='utf-8')
file_handler.setFormatter(formatter)
# ルートロガーの設定
root_logger = logging.getLogger()
root_logger.addHandler(console_handler)
root_logger.addHandler(file_handler)
root_logger.setLevel(logging.INFO)
def _setup_hooks(self):
"""Flaskフックの設定"""
@self.app.before_request
def before_request():
# リクエストIDの生成と設定
request_id = request.headers.get('X-Request-ID') or str(uuid.uuid4())
parent_request_id = request.headers.get('X-Parent-Request-ID')
request_id_var.set(request_id)
parent_request_id_var.set(parent_request_id)
# スレッドローカルストレージに保存
g.request_id = request_id
threading.local().service_context = {
'request_id': request_id,
'parent_request_id': parent_request_id,
'service': 'flask-app',
'endpoint': request.endpoint
}
# リクエスト開始ログ
self.app.logger.info(
'リクエスト開始',
extra={
'extra_fields': {
'http': {
'method': request.method,
'path': request.path,
'query': dict(request.args)
}
}
}
)
@self.app.after_request
def after_request(response):
# レスポンスヘッダーにトレース情報を追加
response.headers['X-Request-ID'] = g.request_id
response.headers['X-Trace-Service'] = 'flask-app'
# リクエスト完了ログ
duration = time.time() - request.start_time if hasattr(request, 'start_time') else 0
self.app.logger.info(
'リクエスト完了',
extra={
'extra_fields': {
'http': {
'status_code': response.status_code,
'duration_ms': int(duration * 1000)
}
}
}
)
return response
# リクエスト開始時間を記録
@self.app.before_request
def record_start_time():
request.start_time = time.time()
def _init_metrics(self):
"""Prometheusメトリクスの初期化"""
# メトリクスストレージ
self.metrics = {
'request_count': 0,
'error_count': 0,
'request_duration': [],
'external_calls': []
}
def trace_external_call(self, service_name, endpoint, method='GET'):
"""外部サービス呼び出しをトレースするデコレータ"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
call_id = str(uuid.uuid4())[:8]
start_time = time.time()
# 呼び出し前ログ
self.app.logger.info(
f'外部サービス呼び出し開始: {service_name}',
extra={
'extra_fields': {
'external_call': {
'id': call_id,
'service': service_name,
'endpoint': endpoint,
'method': method,
'status': 'started'
}
}
}
)
try:
# 関数実行
result = func(*args, **kwargs)
status = 'success'
return result
except Exception as e:
status = 'error'
raise
finally:
# 呼び出し完了ログ
duration = time.time() - start_time
self.app.logger.info(
f'外部サービス呼び出し完了: {service_name}',
extra={
'extra_fields': {
'external_call': {
'id': call_id,
'service': service_name,
'endpoint': endpoint,
'method': method,
'status': status,
'duration_ms': int(duration * 1000)
}
}
}
)
# メトリクスの記録
self.metrics['external_calls'].append({
'service': service_name,
'duration': duration,
'status': status,
'timestamp': datetime.utcnow().isoformat()
})
return wrapper
return decorator
# Flaskアプリケーションの作成
app = Flask(__name__)
logging_system = DistributedLoggingSystem(app)
# トレーシング対応のrequestsラッパー
class TracedRequests:
"""トレーシング対応のHTTPクライアント"""
@staticmethod
@logging_system.trace_external_call('external-api', '/data')
def get_external_data():
"""外部APIを呼び出す（模擬）"""
# 実際のAPI呼び出し
# response = requests.get('https://api.example.com/data', headers={
# 'X-Request-ID': request_id_var.get(),
# 'X-Parent-Request-ID': parent_request_id_var.get()
# })
# return response.json()
# 模擬実装
time.sleep(0.1) # 100msの遅延を模擬
return {'data': 'external_data'}
@app.route('/')
def index():
app.logger.info('インデックスページアクセス')
return jsonify({
'service': 'flask-app',
'request_id': g.request_id,
'timestamp': datetime.utcnow().isoformat()
})
@app.route('/external-call')
def external_call():
"""外部サービスを呼び出すエンドポイント"""
data = TracedRequests.get_external_data()
return jsonify({
'status': 'success',
'data': data,
'request_id': g.request_id
})
@app.route('/metrics')
def metrics():
"""Prometheus形式のメトリクスを返す"""
metrics_output = []
# リクエスト数
metrics_output.append(f'# HELP flask_requests_total Total number of requests')
metrics_output.append(f'# TYPE flask_requests_total counter')
metrics_output.append(f'flask_requests_total {logging_system.metrics["request_count"]}')
# エラー数
metrics_output.append(f'# HELP flask_errors_total Total number of errors')
metrics_output.append(f'# TYPE flask_errors_total counter')
metrics_output.append(f'flask_errors_total {logging_system.metrics["error_count"]}')
return '\n'.join(metrics_output), 200, {'Content-Type': 'text/plain'}
if __name__ == '__main__':
app.run(debug=True, port=5000)上級3: リアルタイムデバッグ・モニタリングダッシュボード
app.py
import json
import time
import psutil
import threading
import logging
from datetime import datetime
from collections import deque
from functools import wraps
from flask import Flask, render_template, request, jsonify
from flask_socketio import SocketIO, emit
import sqlite3
from contextlib import contextmanager
app = Flask(__name__)
app.config['SECRET_KEY'] = 'dashboard-secret-key'
socketio = SocketIO(app, cors_allowed_origins="*")
# モニタリングデータのストレージ
class MonitoringData:
def __init__(self, max_records=1000):
self.logs = deque(maxlen=max_records)
self.requests = deque(maxlen=max_records)
self.metrics = {
'cpu_usage': deque(maxlen=100),
'memory_usage': deque(maxlen=100),
'active_sessions': 0
}
self.sql_queries = deque(maxlen=500)
self.active_traces = {}
# ログキャプチャーの設定
self._setup_log_capture()
def _setup_log_capture(self):
"""ログをキャプチャするハンドラを設定"""
class SocketIOLogHandler(logging.Handler):
def __init__(self, monitoring_data):
super().__init__()
self.monitoring_data = monitoring_data
def emit(self, record):
log_entry = {
'timestamp': datetime.now().isoformat(),
'level': record.levelname,
'message': self.format(record),
'module': record.module,
'funcName': record.funcName
}
# ログを保存
self.monitoring_data.logs.append(log_entry)
# WebSocketで送信
socketio.emit('new_log', log_entry)
# ハンドラの追加
handler = SocketIOLogHandler(self)
handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
root_logger = logging.getLogger()
root_logger.addHandler(handler)
root_logger.setLevel(logging.INFO)
def add_request(self, request_data):
"""リクエストを記録"""
self.requests.append(request_data)
socketio.emit('new_request', request_data)
def add_sql_query(self, query_info):
"""SQLクエリを記録"""
self.sql_queries.append(query_info)
socketio.emit('new_sql_query', query_info)
def update_metrics(self):
"""システムメトリクスを更新"""
metrics = {
'timestamp': datetime.now().isoformat(),
'cpu_percent': psutil.cpu_percent(),
'memory_percent': psutil.virtual_memory().percent,
'memory_used': psutil.virtual_memory().used / 1024 / 1024, # MB
'memory_total': psutil.virtual_memory().total / 1024 / 1024, # MB
'active_threads': threading.active_count(),
'disk_usage': psutil.disk_usage('/').percent
}
# メトリクスを保存
for key in ['cpu_percent', 'memory_percent']:
self.metrics[key].append(metrics[key])
# WebSocketで送信
socketio.emit('metrics_update', metrics)
return metrics
# モニタリングデータのインスタンス
monitoring = MonitoringData()
# SQLクエリトレーサー
def trace_sql_queries(f):
"""SQLクエリをトレースするデコレータ"""
@wraps(f)
def decorated_function(*args, **kwargs):
start_time = time.time()
# 実際のアプリケーションではここでSQLクエリをインターセプト
# この例では模擬実装
# モッククエリ情報
query_info = {
'timestamp': datetime.now().isoformat(),
'query': 'SELECT * FROM users WHERE id = ?',
'params': [1],
'duration': 0.05, # 50ms
'endpoint': request.endpoint if request else 'unknown'
}
# クエリを記録
monitoring.add_sql_query(query_info)
return f(*args, **kwargs)
return decorated_function
# リクエストトレーサー
@app.before_request
def trace_request():
"""リクエストをトレース"""
if request.endpoint and request.endpoint != 'static':
request.start_time = time.time()
# トレースIDの生成
request.trace_id = f"trace_{int(time.time() * 1000)}"
monitoring.active_traces[request.trace_id] = {
'start_time': request.start_time,
'method': request.method,
'path': request.path,
'endpoint': request.endpoint
}
@app.after_request
def trace_response(response):
"""レスポンスをトレース"""
if hasattr(request, 'start_time'):
duration = time.time() - request.start_time
request_data = {
'trace_id': request.trace_id,
'timestamp': datetime.now().isoformat(),
'method': request.method,
'path': request.path,
'endpoint': request.endpoint,
'status_code': response.status_code,
'duration_ms': round(duration * 1000, 2),
'content_length': response.content_length or 0,
'user_agent': request.user_agent.string[:100] if request.user_agent else ''
}
# リクエストを記録
monitoring.add_request(request_data)
# アクティブトレースから削除
if request.trace_id in monitoring.active_traces:
del monitoring.active_traces[request.trace_id]
return response
# メトリクス収集スレッド
def collect_metrics():
"""定期的にメトリクスを収集"""
while True:
monitoring.update_metrics()
time.sleep(2) # 2秒間隔
# メトリクス収集スレッドの開始
metrics_thread = threading.Thread(target=collect_metrics, daemon=True)
metrics_thread.start()
# WebSocketイベント
@socketio.on('connect')
def handle_connect():
"""クライアント接続時の処理"""
app.logger.info('ダッシュボードクライアントが接続しました')
# 現在のデータを送信
emit('initial_data', {
'logs': list(monitoring.logs)[-100:], # 直近100件
'requests': list(monitoring.requests)[-50:], # 直近50件
'sql_queries': list(monitoring.sql_queries)[-20:], # 直近20件
'active_traces': monitoring.active_traces,
'current_metrics': monitoring.update_metrics() # 現在のメトリクスも更新
})
@socketio.on('change_log_level')
def handle_change_log_level(data):
"""ログレベルを変更"""
try:
level_name = data.get('level', 'INFO').upper()
level = getattr(logging, level_name, logging.INFO)
root_logger = logging.getLogger()
root_logger.setLevel(level)
emit('log_level_changed', {
'success': True,
'new_level': level_name,
'message': f'ログレベルを {level_name} に変更しました'
})
app.logger.info(f'ログレベルが {level_name} に変更されました')
except Exception as e:
emit('log_level_changed', {
'success': False,
'message': f'ログレベル変更に失敗: {str(e)}'
})
@socketio.on('clear_data')
def handle_clear_data(data):
"""データをクリア"""
data_type = data.get('type', 'all')
if data_type == 'logs' or data_type == 'all':
monitoring.logs.clear()
if data_type == 'requests' or data_type == 'all':
monitoring.requests.clear()
if data_type == 'sql_queries' or data_type == 'all':
monitoring.sql_queries.clear()
emit('data_cleared', {
'type': data_type,
'message': f'{data_type} のデータをクリアしました'
})
# HTTPルート
@app.route('/')
def dashboard():
"""ダッシュボードメインページ"""
return render_template('dashboard.html')
@app.route('/api/metrics')
def get_metrics():
"""メトリクスAPI"""
return jsonify({
'logs_count': len(monitoring.logs),
'requests_count': len(monitoring.requests),
'sql_queries_count': len(monitoring.sql_queries),
'active_traces_count': len(monitoring.active_traces),
'current_metrics': monitoring.update_metrics(),
'cpu_history': list(monitoring.metrics['cpu_usage']),
'memory_history': list(monitoring.metrics['memory_usage'])
})
@app.route('/api/logs')
def get_logs():
"""ログAPI"""
level = request.args.get('level', 'all')
limit = int(request.args.get('limit', 100))
logs = list(monitoring.logs)
if level != 'all':
logs = [log for log in logs if log['level'] == level.upper()]
return jsonify(logs[-limit:])
@app.route('/api/sql/explain/')
@trace_sql_queries
def explain_query(query_id):
"""SQLクエリの実行計画を取得（模擬）"""
# 実際のアプリケーションでは、ここで実際のSQL EXPLAINを実行
explain_result = {
'query_id': query_id,
'explanation': [
{'id': 1, 'select_type': 'SIMPLE', 'table': 'users', 'type': 'ALL', 'rows': 1000},
{'id': 1, 'select_type': 'SIMPLE', 'table': 'orders', 'type': 'ref', 'rows': 10}
],
'warnings': [
'Using where',
'Using temporary'
],
'estimated_cost': 150.25
}
return jsonify(explain_result)
@app.route('/test/error')
def test_error():
"""エラー発生テストエンドポイント"""
try:
# 意図的にエラーを発生
result = 1 / 0
except Exception as e:
app.logger.error('ゼロ除算エラーが発生しました', exc_info=True)
return 'エラーが発生しました（これはテストです）', 500
return 'これは表示されません'
@app.route('/test/slow')
def test_slow():
"""遅い処理のテストエンドポイント"""
time.sleep(3) # 3秒待機
return '遅い処理が完了しました'
@app.route('/test/sql')
@trace_sql_queries
def test_sql():
"""SQLクエリのテストエンドポイント"""
# 模擬SQLクエリ
queries = [
"SELECT * FROM users WHERE active = 1",
"UPDATE users SET last_login = NOW() WHERE id = 1",
"SELECT COUNT(*) FROM orders WHERE status = 'completed'"
]
for query in queries:
# 各クエリを記録
monitoring.add_sql_query({
'timestamp': datetime.now().isoformat(),
'query': query,
'duration': 0.1,
'endpoint': 'test_sql'
})
return 'SQLクエリテストが完了しました'
if __name__ == '__main__':
socketio.run(app, debug=True, port=5001) templates/dashboard.html
<!DOCTYPE html>
<html>
<head>
<title>Flask リアルタイムデバッグダッシュボード</title>
<script src="https://cdn.socket.io/4.5.0/socket.io.min.js"></script>
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
<style>
* { box-sizing: border-box; margin: 0; padding: 0; }
body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: #f5f5f5; color: #333; }
.container { max-width: 100%; padding: 20px; }
.header { background: #2c3e50; color: white; padding: 20px; margin-bottom: 20px;
border-radius: 8px; }
.grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(300px, 1fr));
gap: 20px; margin-bottom: 20px; }
.card { background: white; border-radius: 8px; padding: 20px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }
.card h3 { margin-bottom: 15px; color: #2c3e50; border-bottom: 2px solid #3498db;
padding-bottom: 8px; }
.metrics { display: grid; grid-template-columns: repeat(auto-fit, minmax(150px, 1fr));
gap: 10px; }
.metric-item { text-align: center; padding: 15px; background: #f8f9fa;
border-radius: 6px; }
.metric-value { font-size: 24px; font-weight: bold; color: #3498db; }
.metric-label { font-size: 12px; color: #666; margin-top: 5px; }
.log-entry { padding: 8px; border-bottom: 1px solid #eee; font-size: 12px; }
.log-entry.error { background: #ffe6e6; }
.log-entry.warning { background: #fff3cd; }
.log-entry.info { background: #e6f7ff; }
.controls { display: flex; gap: 10px; margin-bottom: 20px; }
button, select { padding: 10px 15px; border: none; border-radius: 4px;
cursor: pointer; font-size: 14px; }
button.primary { background: #3498db; color: white; }
button.danger { background: #e74c3c; color: white; }
select { background: white; border: 1px solid #ddd; }
.chart-container { position: relative; height: 200px; }
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>Flask リアルタイムデバッグダッシュボード</h1>
<p>アプリケーションの状態をリアルタイムで監視</p>
</div>
<div class="controls">
<select id="logLevel">
<option value="DEBUG">DEBUG</option>
<option value="INFO" selected>INFO</option>
<option value="WARNING">WARNING</option>
<option value="ERROR">ERROR</option>
<option value="CRITICAL">CRITICAL</option>
</select>
<button class="primary" onclick="changeLogLevel()">ログレベル変更</button>
<button onclick="clearData('logs')">ログクリア</button>
<button onclick="clearData('requests')">リクエストクリア</button>
<button class="danger" onclick="clearData('all')">全データクリア</button>
</div>
<div class="grid">
<div class="card">
<h3>システムメトリクス</h3>
<div class="metrics">
<div class="metric-item">
<div class="metric-value" id="cpuPercent">0%</div>
<div class="metric-label">CPU使用率</div>
</div>
<div class="metric-item">
<div class="metric-value" id="memoryPercent">0%</div>
<div class="metric-label">メモリ使用率</div>
</div>
<div class="metric-item">
<div class="metric-value" id="activeThreads">0</div>
<div class="metric-label">アクティブスレッド</div>
</div>
<div class="metric-item">
<div class="metric-value" id="activeTraces">0</div>
<div class="metric-label">アクティブトレース</div>
</div>
</div>
<div class="chart-container">
<canvas id="metricsChart"></canvas>
</div>
</div>
<div class="card">
<h3>リクエスト統計</h3>
<div class="metrics">
<div class="metric-item">
<div class="metric-value" id="totalRequests">0</div>
<div class="metric-label">総リクエスト数</div>
</div>
<div class="metric-item">
<div class="metric-value" id="avgResponseTime">0ms</div>
<div class="metric-label">平均応答時間</div>
</div>
<div class="metric-item">
<div class="metric-value" id="errorRate">0%</div>
<div class="metric-label">エラー率</div>
</div>
<div class="metric-item">
<div class="metric-value" id="activeSessions">0</div>
<div class="metric-label">アクティブセッション</div>
</div>
</div>
</div>
</div>
<div class="grid">
<div class="card">
<h3>リアルタイムログ</h3>
<div id="logContainer" style="max-height: 300px; overflow-y: auto;">
<!-- ログがここに表示されます -->
</div>
</div>
<div class="card">
<h3>最近のリクエスト</h3>
<div id="requestContainer" style="max-height: 300px; overflow-y: auto;">
<!-- リクエストがここに表示されます -->
</div>
</div>
</div>
<div class="card">
<h3>SQLクエリパフォーマンス</h3>
<div id="sqlContainer" style="max-height: 200px; overflow-y: auto;">
<!-- SQLクエリがここに表示されます -->
</div>
</div>
</div>
<script>
// Socket.IO接続
const socket = io();
// チャートの初期化
const metricsCtx = document.getElementById('metricsChart').getContext('2d');
const metricsChart = new Chart(metricsCtx, {
type: 'line',
data: {
labels: [],
datasets: [
{
label: 'CPU使用率 (%)',
data: [],
borderColor: '#3498db',
backgroundColor: 'rgba(52, 152, 219, 0.1)',
tension: 0.4
},
{
label: 'メモリ使用率 (%)',
data: [],
borderColor: '#2ecc71',
backgroundColor: 'rgba(46, 204, 113, 0.1)',
tension: 0.4
}
]
},
options: {
responsive: true,
maintainAspectRatio: false,
plugins: {
legend: { position: 'top' }
},
scales: {
y: {
beginAtZero: true,
max: 100
}
}
}
});
// 初期データの受信
socket.on('initial_data', function(data) {
updateMetrics(data.current_metrics);
updateLogs(data.logs);
updateRequests(data.requests);
updateSqlQueries(data.sql_queries);
});
// 新しいログの受信
socket.on('new_log', function(log) {
addLogEntry(log);
});
// 新しいリクエストの受信
socket.on('new_request', function(request) {
addRequestEntry(request);
});
// 新しいSQLクエリの受信
socket.on('new_sql_query', function(query) {
addSqlQuery(query);
});
// メトリクス更新の受信
socket.on('metrics_update', function(metrics) {
updateMetrics(metrics);
});
// ログレベル変更結果の受信
socket.on('log_level_changed', function(result) {
alert(result.message);
});
// データクリア結果の受信
socket.on('data_cleared', function(result) {
alert(result.message);
if (result.type === 'logs' || result.type === 'all') {
document.getElementById('logContainer').innerHTML = '';
}
if (result.type === 'requests' || result.type === 'all') {
document.getElementById('requestContainer').innerHTML = '';
}
});
// メトリクス更新関数
function updateMetrics(metrics) {
document.getElementById('cpuPercent').textContent = metrics.cpu_percent.toFixed(1) + '%';
document.getElementById('memoryPercent').textContent = metrics.memory_percent.toFixed(1) + '%';
document.getElementById('activeThreads').textContent = metrics.active_threads;
document.getElementById('activeTraces').textContent = Object.keys(window.activeTraces || {}).length;
// チャートデータの更新
const timestamp = new Date(metrics.timestamp).toLocaleTimeString();
if (metricsChart.data.labels.length >= 20) {
metricsChart.data.labels.shift();
metricsChart.data.datasets[0].data.shift();
metricsChart.data.datasets[1].data.shift();
}
metricsChart.data.labels.push(timestamp);
metricsChart.data.datasets[0].data.push(metrics.cpu_percent);
metricsChart.data.datasets[1].data.push(metrics.memory_percent);
metricsChart.update();
}
// ログ追加関数
function addLogEntry(log) {
const container = document.getElementById('logContainer');
const logEntry = document.createElement('div');
logEntry.className = `log-entry ${log.level.toLowerCase()}`;
logEntry.innerHTML = `
<strong>[${log.timestamp.split('T')[1].slice(0,8)}]</strong>
<span class="level">${log.level}</span>
<span>${log.message}</span>
`;
container.appendChild(logEntry);
container.scrollTop = container.scrollHeight;
// ログが多すぎる場合は古いものを削除
if (container.children.length > 100) {
container.removeChild(container.firstChild);
}
}
// 初期ログ更新関数
function updateLogs(logs) {
const container = document.getElementById('logContainer');
container.innerHTML = '';
logs.forEach(log => {
addLogEntry(log);
});
}
// リクエスト追加関数
function addRequestEntry(request) {
const container = document.getElementById('requestContainer');
const requestEntry = document.createElement('div');
requestEntry.className = 'log-entry';
requestEntry.innerHTML = `
<strong>${request.method} ${request.path}</strong>
<span style="float: right; color: ${request.status_code >= 400 ? '#e74c3c' : '#27ae60'}">
${request.status_code} (${request.duration_ms}ms)
</span>
<br>
<small>${request.timestamp.split('T')[1].slice(0,8)}</small>
`;
container.appendChild(requestEntry);
container.scrollTop = container.scrollHeight;
// リクエストが多すぎる場合は古いものを削除
if (container.children.length > 50) {
container.removeChild(container.firstChild);
}
}
// 初期リクエスト更新関数
function updateRequests(requests) {
const container = document.getElementById('requestContainer');
container.innerHTML = '';
requests.forEach(request => {
addRequestEntry(request);
});
}
// SQLクエリ追加関数
function addSqlQuery(query) {
const container = document.getElementById('sqlContainer');
const queryEntry = document.createElement('div');
queryEntry.className = 'log-entry';
queryEntry.innerHTML = `
<strong>${query.endpoint}</strong>
<span style="float: right; color: #f39c12">
${(query.duration * 1000).toFixed(1)}ms
</span>
<br>
<small>${query.query}</small>
`;
container.appendChild(queryEntry);
container.scrollTop = container.scrollHeight;
}
// 初期SQLクエリ更新関数
function updateSqlQueries(queries) {
const container = document.getElementById('sqlContainer');
container.innerHTML = '';
queries.forEach(query => {
addSqlQuery(query);
});
}
// ログレベル変更関数
function changeLogLevel() {
const level = document.getElementById('logLevel').value;
socket.emit('change_log_level', { level: level });
}
// データクリア関数
function clearData(type) {
socket.emit('clear_data', { type: type });
}
// 定期的に統計を更新
setInterval(() => {
fetch('/api/metrics')
.then(response => response.json())
.then(data => {
document.getElementById('totalRequests').textContent = data.requests_count;
document.getElementById('activeSessions').textContent = data.active_traces_count;
});
}, 5000);
</script>
</body>
</html>同じカテゴリーの記事
Flaskでのフォーム処理の基礎
はじめに Webアプリケーションにおいて、フォームはユーザーからの入力を受け付けるための重要な要素です。Flaskでは、シンプルかつ効果的な方法でフォーム処理を […]
FlaskテンプレートエンジンJinja2の基礎
はじめに 前回までにFlaskの環境構築とルーティングについて学びました。今回は、Webアプリケーションの見た目を担当する「テンプレートエンジンJinja2」に […]
Flaskルーティングとビュー関数
はじめに 前回の記事でFlask開発環境の構築を完了したら、次はWebアプリケーションの核心である「ルーティング」と「ビュー関数」について学びましょう。ルーティ […]