Flask のシグナルは、アプリケーション内で特定のイベントが発生したときに通知を受け取る仕組みである。blinker ライブラリをベースにしており、疎結合なイベント駆動設計を実現できる。
シグナルの基本
Flask には組み込みのシグナルがいくつか用意されている。
from flask import Flask, request_started, request_finished app = Flask(__name__) @request_started.connect_via(app) def on_request_started(sender, **kwargs): print('Request started') @request_finished.connect_via(app) def on_request_finished(sender, response, **kwargs): print(f'Request finished with status {response.status_code}')
組み込みシグナル一覧
| request_started | リクエスト処理開始時 |
| request_finished | リクエスト処理完了時 |
| request_tearing_down | リクエストコンテキスト破棄時 |
| got_request_exception | 未処理の例外発生時 |
| appcontext_pushed | アプリケーションコンテキストがプッシュされた時 |
| appcontext_popped | アプリケーションコンテキストがポップされた時 |
カスタムシグナルの作成
アプリケーション固有のイベントに対してシグナルを定義できる。
from blinker import Namespace app_signals = Namespace() user_logged_in = app_signals.signal('user-logged-in') order_created = app_signals.signal('order-created')
シグナルの発火
from flask import g @app.route('/login', methods=['POST']) def login(): user = authenticate(request.form['username'], request.form['password']) if user: session['user_id'] = user.id # シグナルを発火 user_logged_in.send(app, user=user) return 'Logged in' return 'Invalid credentials', 401
シグナルの購読
@user_logged_in.connect_via(app) def on_user_logged_in(sender, user, **kwargs): # ログイン履歴を記録 LoginHistory.create(user_id=user.id, ip=request.remote_addr) @user_logged_in.connect_via(app) def send_welcome_notification(sender, user, **kwargs): # 通知を送信 send_notification(user, 'Welcome back!')
複数のハンドラを同じシグナルに登録できる。処理の順序は保証されない。
シグナルの利点
コードの疎結合化（ログイン処理とログ記録を分離）
プラグインアーキテクチャの実現
テスト時にシグナルをモックできる
注意点
シグナルは同期的に実行される。重い処理はバックグラウンドタスクに委譲すべきである。
@order_created.connect_via(app) def on_order_created(sender, order, **kwargs): # 重い処理は Celery などに委譲 send_confirmation_email.delay(order.id)
また、シグナルハンドラ内で例外が発生すると、他のハンドラが実行されない可能性がある。適切な例外処理を入れること。
