信号

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

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