Flask のシグナルを使ったイベント駆動設計

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)

また、シグナルハンドラ内で例外が発生すると、他のハンドラが実行されない可能性がある。適切な例外処理を入れること。