訊號

訊號是一種輕量級的方式,用於在應用程式和每個請求的生命週期中通知訂閱者某些事件。當事件發生時,它會發射訊號,進而呼叫每個訂閱者。

訊號由 Blinker 函式庫實作。請參閱其文件以取得詳細資訊。Flask 提供了一些內建訊號。擴充套件可以提供它們自己的訊號。

許多訊號反映了 Flask 基於裝飾器的回呼,名稱相似。例如,request_started 訊號類似於 before_request() 裝飾器。訊號相較於處理器的優勢在於它們可以被臨時訂閱,並且不能直接影響應用程式。這對於測試、指標、稽核等非常有用。例如,如果您想知道在請求的哪些部分渲染了哪些範本,有一個訊號會通知您該資訊。

核心訊號

請參閱 訊號 以取得所有內建訊號的列表。《應用程式結構與生命週期》頁面也描述了訊號和裝飾器的執行順序。

訂閱訊號

要訂閱訊號,您可以使用訊號的 connect() 方法。第一個參數是當訊號發射時應該呼叫的函式,可選的第二個參數指定發送者。要取消訂閱訊號,您可以使用 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 變數中。每當渲染範本時,範本物件和上下文都會附加到其中。

此外,還有一個方便的輔助方法 (connected_to()),它允許您使用上下文管理器將函式臨時訂閱到訊號。由於無法以這種方式指定上下文管理器的傳回值,因此您必須將列表作為參數傳遞進去

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]

建立訊號

如果您想在自己的應用程式中使用訊號,可以直接使用 blinker 函式庫。最常見的用例是在自訂 Namespace 中的命名訊號。這是大多數時候推薦的做法

from blinker import Namespace
my_signals = Namespace()

現在您可以像這樣建立新的訊號

model_saved = my_signals.signal('model-saved')

此處訊號的名稱使其獨一無二,並且簡化了除錯。您可以使用 name 屬性存取訊號的名稱。

發送訊號

如果您想發射訊號,可以透過呼叫 send() 方法來實現。它接受發送者作為第一個參數,並可選地接受一些關鍵字參數,這些參數將轉發給訊號訂閱者

class Model(object):
    ...

    def save(self):
        model_saved.send(self)

盡量總是選擇一個好的發送者。如果您的類別正在發射訊號,請將 self 作為發送者傳遞。如果您從隨機函式發射訊號,您可以將 current_app._get_current_object() 作為發送者傳遞。

將代理作為發送者傳遞

永遠不要將 current_app 作為發送者傳遞給訊號。請改用 current_app._get_current_object()。這樣做的原因是 current_app 是一個代理,而不是真正的應用程式物件。

訊號和 Flask 的請求上下文

訊號在接收訊號時完全支援請求上下文。《請求上下文》頁面也描述了訊號和裝飾器的執行順序。上下文本地變數在 request_startedrequest_finished 之間始終可用,因此您可以根據需要依賴 flask.g 和其他變數。請注意發送訊號request_tearing_down 訊號中描述的限制。

基於裝飾器的訊號訂閱

您也可以透過使用 connect_via() 裝飾器輕鬆訂閱訊號

from flask import template_rendered

@template_rendered.connect_via(app)
def when_template_rendered(sender, template, context, **extra):
    print(f'Template {template.name} is rendered with {context}')