信號(hào)(signals)

Flask信號(hào)(signals, or event hooking)允許特定的發(fā)送端通知訂閱者發(fā)生了什么（既然知道發(fā)生了什么，那我們可以知道接下來(lái)該做什么了）。

Flask提供了一些信號(hào)（核心信號(hào)）且其它的擴(kuò)展提供更多的信號(hào)。信號(hào)是用于通知訂閱者，而不應(yīng)該鼓勵(lì)訂閱者修改數(shù)據(jù)。相關(guān)信號(hào)請(qǐng)查閱文檔。

信號(hào)依賴于Blinker庫(kù)。

鉤子(hooks)

Flask鉤子（通常出現(xiàn)在藍(lán)圖或應(yīng)用程序現(xiàn)存的方法中，比如一些內(nèi)置裝飾器，例如before_request）不需要Blinker庫(kù)并且允許你改變請(qǐng)求對(duì)象(request)或者響應(yīng)對(duì)象(response)。這些改變了應(yīng)用程序（或者藍(lán)圖）的行為。比如before_request()裝飾器。

信號(hào)看起來(lái)和鉤子做同樣的事情。然而在工作方式上它們存在不同。譬如核心的before_request()處理程序以特定的順序執(zhí)行，并且可以在返回響應(yīng)之前放棄請(qǐng)求。相比之下，所有的信號(hào)處理器是無(wú)序執(zhí)行的，并且不修改任何數(shù)據(jù)。

一般來(lái)說(shuō)，鉤子用于改變行為（比如，身份驗(yàn)證或錯(cuò)誤處理），而信號(hào)用于記錄事件（比如記錄日志）。

創(chuàng)建信號(hào)

因?yàn)樾盘?hào)依賴于Blinker庫(kù)，請(qǐng)確保已經(jīng)安裝。

如果你要在自己的應(yīng)用中使用信號(hào)，你可以直接使用Blinker庫(kù)。最常見(jiàn)的使用情況是命名一個(gè)自定義的Namespace的信號(hào)。這也是大多數(shù)時(shí)候推薦的做法:

代碼如下:

from blinker import Namespace

my_signals = Namespace()

現(xiàn)在你可以像這樣創(chuàng)建新的信號(hào):

代碼如下:

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

這里使用唯一的信號(hào)名并且簡(jiǎn)化調(diào)試。可以用name屬性來(lái)訪問(wèn)信號(hào)名。

對(duì)擴(kuò)展開(kāi)發(fā)者:

如果你正在編寫(xiě)一個(gè)Flask擴(kuò)展，你想優(yōu)雅地減少缺少Blinker安裝的影響，你可以這樣做使用flask.signals.Namespace類。

訂閱信號(hào)

你可以使用信號(hào)的connect()方法來(lái)訂閱信號(hào)。第一個(gè)參數(shù)是信號(hào)發(fā)出時(shí)要調(diào)用的函數(shù)，第二個(gè)參數(shù)是可選的，用于確定信號(hào)的發(fā)送者。一個(gè)信號(hào)可以擁有多個(gè)訂閱者。你可以用disconnect()方法來(lái)退訂信號(hào)。

對(duì)于所有的核心Flask信號(hào)，發(fā)送者都是發(fā)出信號(hào)的應(yīng)用。當(dāng)你訂閱一個(gè)信號(hào)，請(qǐng)確保也提供一個(gè)發(fā)送者，除非你確實(shí)想監(jiān)聽(tīng)全部應(yīng)用的信號(hào)。這在你開(kāi)發(fā)一個(gè)擴(kuò)展的時(shí)候尤其正確。

比如這里有一個(gè)用于在單元測(cè)試中找出哪個(gè)模板被渲染和傳入模板的變量的助手上下文管理器:

代碼如下:

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)

現(xiàn)在可以輕松地與測(cè)試客戶端配對(duì):

代碼如下:

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

確保訂閱使用了一個(gè)額外的 **extra 參數(shù)，這樣當(dāng) Flask 對(duì)信號(hào)引入新參數(shù)時(shí)你的調(diào)用不會(huì)失敗。

代碼中，從 with 塊的應(yīng)用 app 中流出的渲染的所有模板現(xiàn)在會(huì)被記錄到templates 變量。無(wú)論何時(shí)模板被渲染，模板對(duì)象的上下文中添加上它。

此外，也有一個(gè)方便的助手方法(connected_to()) ，它允許你臨時(shí)地用它自己的上下文管理器把函數(shù)訂閱到信號(hào)。因?yàn)樯舷挛墓芾砥鞯姆祷刂挡荒馨茨欠N方法給定，則需要把列表作為參數(shù)傳入:

代碼如下:

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)

上面的例子看起來(lái)像這樣:

代碼如下:

templates = []

with captured_templates(app, templates, **extra):

    ...

    template, context = templates[0]

發(fā)送信號(hào)

如果你想要發(fā)送信號(hào)，你可以通過(guò)調(diào)用 send() 方法來(lái)達(dá)到目的。它接受一個(gè)發(fā)件者作為第一個(gè)參數(shù)和一些可選的被轉(zhuǎn)發(fā)到信號(hào)用戶的關(guān)鍵字參數(shù):

代碼如下:

class Model(object):

    ...

    def save(self):

        model_saved.send(self)

永遠(yuǎn)嘗試選擇一個(gè)合適的發(fā)送者。如果你有一個(gè)發(fā)出信號(hào)的類，把 self 作為發(fā)送者。如果你從一個(gè)隨機(jī)的函數(shù)發(fā)出信號(hào)，把current_app._get_current_object()作為發(fā)送者。

基于信號(hào)訂閱的裝飾器

在Blinker 1.1中通過(guò)使用新的connect_via()裝飾器你也能夠輕易地訂閱信號(hào):

代碼如下:

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信號(hào)是Flask核心信號(hào)。

當(dāng)一個(gè)模版成功地渲染，這個(gè)信號(hào)會(huì)被發(fā)送。這個(gè)信號(hào)與模板實(shí)例 template 和上下文的詞典（名為 context ）一起調(diào)用。

信號(hào)發(fā)送：

代碼如下:

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)

帳號(hào)追蹤

user_logged_in是Flask-User中定義的信號(hào)，當(dāng)用戶成功登入之后，這個(gè)信號(hào)會(huì)被發(fā)送。

發(fā)送信號(hào)：

代碼如下:

# Send user_logged_in signal

user_logged_in.send(current_app._get_current_object(), user=user)

下面這個(gè)例子追蹤登錄次數(shù)和登錄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()

小結(jié)

信號(hào)可以讓你在一瞬間安全地訂閱它們。例如，這些臨時(shí)的訂閱對(duì)測(cè)試很有幫助。使用信號(hào)時(shí)，不要讓信號(hào)訂閱者（接收者）發(fā)生異常，因?yàn)楫惓?huì)造成程序中斷。

更多信息請(qǐng)查看IT技術(shù)專欄