信号

信号是一种轻量级的方式,用于在应用程序和每个请求的生命周期中通知订阅者某些事件。当事件发生时,它会发出信号,从而调用每个订阅者。

信号由 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}')