訊號¶
訊號是一種輕量級的方式,用於在應用程式和每個請求的生命週期中通知訂閱者某些事件。當事件發生時,它會發射訊號,進而呼叫每個訂閱者。
訊號由 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_started
和 request_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}')