Flask 提供了訊號(Signals)功能,是一種訊息分發機制。類似於鉤子(Hooks)。使用訊號功能可以降低程式的耦合,分解複雜的業務模型。例如更新了產品資料後,可以發送一個訊號。當有需要對產品資料進行處理的功能時,就可以捕捉訊號進行處理。例如要建立產品緩存,或是更新搜尋索引等。
定義訊號
Flask 訊號功能使用了Blinker 模組,所以需要先安裝Blinker 模組
pip install blinker
定義一個訊號:
from blinker import Namespace product_saved = Namespace()
也可以使用Flask 包裝的singles 物件:
from flask.singles import Namespace
傳送訊號
傳送訊號需要帶app 實例方法,範例如下:
product_saved.send(app, product=product)
app 後面可以加入要傳遞的參數,但必須以name=value 的格式,不支援使用單一變數名稱的方式。
收接訊號
接收訊號可以使用connect_via 裝飾器函數:
@product_saved.connect_via(app) def updateCache(app, product): print(product)
##Flask 中有以下核心訊號:
1.flask.template_rendered
這個訊號發送於一個模板被渲染成功後。訊號傳遞的template是模板的實例,context是環境物件是字典。 訂閱範例: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
這個訊號傳送於請求開始前,且請求環境設定完成之後。因為請求環境已經綁定, 所以訂閱者可以用標準的全域代理,如 request 來操作請求。 訂閱範例: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
這個訊號傳送於請求進行中發生異常的時候。它的發送 早於 標準異常處理介於。 在調試模式下,雖然沒有異常處理,但發生異常時也會發送這個訊號。訊號傳遞的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)
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)
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)