Python の Flask フレームワークにおけるシグナルシグナルメカニズムの詳細な説明-Python チュートリアル-php.cn

Flask は、メッセージ配信メカニズムであるシグナル機能を提供します。フックに似ています。シグナル関数を使用すると、プログラムの結合を減らし、複雑なビジネスモデルを分解できます。たとえば、製品データを更新した後にシグナルを送信できます。製品データを処理する必要がある機能がある場合、信号をキャプチャして処理できます。たとえば、商品キャッシュを作成したり、検索インデックスを更新したりする必要があります。

シグナルを定義

Flask シグナル関数は Blinker モジュールを使用するため、最初に Blinker モジュールをインストールする必要があります

pip install blinker

ログイン後にコピー

シグナルを定義します:

from blinker import Namespace
product_saved = Namespace()

ログイン後にコピー

Flask でラップされたシングルオブジェクトを使用することもできます。

from flask.singles import Namespace

ログイン後にコピー

シグナルを送信

シグナルを送信するには、アプリインスタンスメソッドを使用する必要があります。例は次のとおりです。

product_saved.send(app, product=product)

ログイン後にコピー

後でアプリに渡されるパラメーターを追加できますが、名前=値の形式にする必要があります。単一の変数名の使用はサポートされていません。

信号を受信

connect_via デコレーター関数を使用して信号を受信できます。

@product_saved.connect_via(app)
def updateCache(app, product):
  print(product)

ログイン後にコピー

Flask には次のコアシグナルがあります:

1.flask.template_rendered

このシグナルは、テンプレートが正常にレンダリングされた後に送信されます。シグナルによって渡されるテンプレートはテンプレートのインスタンスであり、コンテキストは辞書である環境オブジェクトです。

サブスクリプションの例:

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)

ログイン後にコピー

2.flask.request_started

このシグナルは、リクエストの開始前とリクエスト環境のセットアップ後に送信されます。リクエストコンテキストはすでにバインドされているため、サブスクライバはリクエストなどの標準グローバルプロキシを使用してリクエストを操作できます。

サブスクリプションの例:

def log_request(sender, **extra):
  sender.logger.debug('Request context is set up')
from flask import request_started
request_started.connect(log_request, app)
flask.request_finished

ログイン後にコピー

このシグナルは、クライアントに応答を送信する前に送信されます。シグナルによって渡される応答が、送信される応答です。

サブスクリプションの例:

def log_response(sender, response, **extra):
  sender.logger.debug('Request context is about to close down. '
            'Response: %s', response)
from flask import request_finished
request_finished.connect(log_response, app)
flask.got_request_exception

ログイン後にコピー

このシグナルは、リクエスト中に例外が発生したときに送信されます。これは、標準の例外処理よりも早く送信されます。デバッグモードでは、例外処理はありませんが、例外が発生したときにもこの信号が送信されます。シグナルによって渡される例外は例外オブジェクトです。

サブスクリプションの例:

def log_exception(sender, exception, **extra):
  sender.logger.debug('Got exception during processing: %s', exception)
from flask import got_request_exception
got_request_exception.connect(log_exception, app)
flask.request_tearing_down

ログイン後にコピー

このシグナルは、例外が発生したかどうかに関係なく、リクエストがクラッシュしたときに送信されます。現在、このシグナルをリッスンする関数は一般的なクラッシュハンドラーの後に呼び出されますが、使用できるものはありません。

サブスクリプションの例:

def close_db_connection(sender, **extra):
  session.close()from flask import appcontext_tearing_down
request_tearing_down.connect(close_db_connection, app)

ログイン後にコピー

Flask バージョン 0.9 では、exc キーワード引数 (存在する場合) も渡します。このパラメータは、クラッシュの原因となった例外への参照です。

3.flask.appcontext_tearing_down

アプリケーション環境がクラッシュしたときにこのシグナルを送信します。このシグナルは、例外によって発生したクラッシュの場合でも、常に送信されます。このシグナルをリッスンする関数は通常のクラッシュハンドラーの後に呼び出されますが、このシグナルに応答することはできません。

サブスクリプションの例:

def close_db_connection(sender, **extra):
  session.close()from flask import request_tearing_down
appcontext_tearing_down.connect(close_db_connection, app)

ログイン後にコピー

これにより、exc キーワード引数も渡されます (存在する場合)。このパラメータは、クラッシュの原因となった例外への参照です。

4.flask.appcontext_pushed

アプリケーションの環境がプッシュされると、アプリケーションはこのシグナルを送信します。この信号は通常、情報を一時的にフックするために単体テストで使用されます。たとえば、g オブジェクト内の既存のリソースを変更するために使用できます。

使用例:

from contextlib import contextmanagerfrom
flask import appcontext_pushed
@contextmanagerdef user_set(app, user):
  def handler(sender, **kwargs):
    g.user = user
  with appcontext_pushed.connected_to(handler, app):
    yield

ログイン後にコピー

テストコードにこれを書きます:

def test_user_me(self):
  with user_set(app, 'john'):
    c = app.test_client()
    resp = c.get('/users/me')
    assert resp.data == 'username=john'
New in version 0.10.

ログイン後にコピー

5.appcontext_popped

アプリケーションの環境がポップアップすると、アプリケーションはこのシグナルを送信します。この信号は通常、appcontext_tearing_down 信号として記述されます。

6.flask.message_flashed

この信号は、アプリがメッセージを点滅させるときに送信されます。 message` パラメータはメッセージの内容、category パラメータはメッセージのカテゴリです。

サブスクリプションの例:

recorded = []def record(sender, message, category, **extra):
  recorded.append((message, category))
from flask import message_flashed
message_flashed.connect(record, app)

ログイン後にコピー

概要
シグナルを使用すると、瞬時に安全に購読できるようになります。これらの一時的なサブスクリプションは、テストなどに役立ちます。シグナルを使用する場合、例外が発生するとプログラムが中断されるため、シグナルサブスクライバ (受信側) で例外が発生しないようにしてください。