Python WebフレームワークFlask信号メカニズム(signals)紹介
Flask信号(signals,or event hooking)は、特定の送信側が購読者に何が起こったかを通知することを許可する(何が起こったかを知っている以上、次に何をすべきかを知ることができる).
Flaskはいくつかの信号(コア信号)を提供し、他の拡張はより多くの信号を提供する.信号は、サブスクライバにデータの変更を奨励すべきではなく、サブスクライバに通知するために使用される.関連信号はドキュメントを参照してください.
信号はBlinkerライブラリに依存します.
フック
Flaskフック(通常、青写真やアプリケーションの既存の方法で表示されます.例えば、before_requestなどの内蔵装飾器はBlinkerライブラリを必要とせず、リクエストオブジェクト(request)やレスポンスオブジェクト(response)を変更することができます.これらは、アプリケーション(または青写真)の動作を変更します.例えばbefore_request()装飾器.
信号はフックと同じように見えます.しかし、仕事のやり方には違いがあります.例えばコアのbefore_request()プロセッサは、特定の順序で実行され、応答を返す前に要求を放棄することができる.対照的に、すべての信号プロセッサは無秩序に実行され、データは変更されません.
一般に、フックは、認証やエラー処理などの動作を変更するために使用され、信号は、ログなどのイベントを記録するために使用される.
シグナルの作成
信号はBlinkerライブラリに依存しますので、インストールされていることを確認してください.
自分のアプリケーションで信号を使用する場合は、Blinkerライブラリを直接使用できます.最も一般的な使用例は、カスタムNamespaceと命名された信号です.これも多くの場合、お勧めの方法です.
from blinker import Namespace
my_signals = Namespace()
このようにして新しい信号を作成できます.
model_saved = my_signals.signal('model-saved')
ここでは、一意の信号名を使用し、デバッグを簡略化します.nameプロパティを使用して信号名にアクセスできます.
拡張開発者:
Flask拡張を作成している場合は、Blinkerのインストールが欠けている影響を優雅に減らしたい場合は、flaskを使用することができます.signals.Namespaceクラス.
サブスクリプションしんごう
信号のconnect()メソッドを使用して信号を購読することができます.1番目のパラメータは、信号の送信者を決定するためにオプションで使用される信号の送信時に呼び出される関数です.1つの信号は、複数の購読者を有することができる.disconnect()メソッドで信号をキャンセルすることができます.
すべてのコアFlask信号に対して、送信者は信号を送信するアプリケーションである.信号を購読する場合は、すべてのアプリケーションの信号を本当に傍受したい場合を除き、送信者も提供してください.これは、拡張を開発するときに特に正しいです.
たとえば、セルテストでテンプレートがレンダリングされ、入力された変数を特定するアシスタントコンテキストマネージャがあります.
from flask import template_rendered
from contextlib import contextmanager
@contextmanager
def captured_templates(app):
recorded = []
def record(sender, template, context, **extra):
recorded.append((template, context))
template_rendered.connect(record, app)
try:
yield recorded
finally:
template_rendered.disconnect(record, app)
テストクライアントと簡単にペアリングできるようになりました.
with captured_templates(app) as templates:
rv = app.test_client().get('/')
assert rv.status_code == 200
assert len(templates) == 1
template, context = templates[0]
assert template.name == 'index.html'
assert len(context['items']) == 10
サブスクリプションに追加の*extraパラメータが使用されていることを確認します.これにより、Flaskが信号に新しいパラメータを導入した場合、呼び出しに失敗しません.
コードでは、withブロックのアプリケーションappから流出したレンダリングのすべてのテンプレートがtemplates変数に記録されます.テンプレートがレンダリングされるたびに、テンプレートオブジェクトのコンテキストに追加されます.
また、独自のコンテキストマネージャで関数を一時的に信号に購読できる便利なアシスタントメソッドもあります.コンテキストマネージャの戻り値は、そのような方法で指定できないため、リストをパラメータとして入力する必要があります.
from flask import template_rendered
def captured_templates(app, recorded, **extra):
def record(sender, template, context):
recorded.append((template, context))
return template_rendered.connected_to(record, app)
上記の例は次のように見えます.
templates = []
with captured_templates(app, templates, **extra):
...
template, context = templates[0]
送信信号
信号を送信したい場合はsend()メソッドを呼び出すことで目的を達成できます.送信者は、第1のパラメータとして、信号ユーザに転送される任意のキーワードパラメータとして受信される.
class Model(object):
...
def save(self):
model_saved.send(self)
適切な送信者を常に選択してみます.信号を送信するクラスがある場合はselfを送信者とします.ランダムな関数から信号を送るとcurrent_app._get_current_object()を送信者とする.
シグナルベースのアクセサリ
Blinker 1.1で新しいconnect_を使用via()装飾器は簡単に信号を購読することができます.
from flask import template_rendered
@template_rendered.connect_via(app)
def when_template_rendered(sender, template, context, **extra):
print 'Template %s is rendered with %s' % (template.name, context)
例を挙げる
テンプレートレンダリング
template_rendered信号はFlaskコア信号である.
テンプレートが正常にレンダリングされると、この信号が送信されます.この信号は、テンプレートインスタンスtemplateおよびコンテキストの辞書(contextという)とともに呼び出される.
信号送信:
def _render(template, context, app):
"""Renders the template and fires the signal"""
rv = template.render(context)
template_rendered.send(app, template=template, context=context)
return rv
サブスクリプションの例:
def log_template_renders(sender, template, context, **extra):
sender.logger.debug('Rendering template "%s" with context %s',
template.name or 'string template',
context)
from flask import template_rendered
template_rendered.connect(log_template_renders, app)
アカウントトラッキング
user_logged_inはFlask-Userで定義された信号であり、ユーザが正常にログインした後、この信号が送信される.
送信信号:
# Send user_logged_in signal
user_logged_in.send(current_app._get_current_object(), user=user)
次の例では、ログイン回数とログインIPを追跡します.
# This code has not been tested
from flask import request
from flask_user.signals import user_logged_in
@user_logged_in.connect_via(app)
def _track_logins(sender, user, **extra):
user.login_count += 1
user.last_login_ip = request.remote_addr
db.session.add(user)
db.session.commit()
小結
信号は一瞬で安全に購読することができます.たとえば、これらの一時的なサブスクリプションはテストに役立ちます.信号を使用する場合は、異常によりプログラムが中断するため、信号購読者(受信者)に異常を起こさせないでください.