概述

信号（Signal）是 Flask 中一个比较鲜为人知的功能，在官方文档中也对此着墨不多。的确，Signal 并不是 Flask 的核心功能————你完全可以在不使用 Signal 的前提下写出完整的 Flask 应用。但在某些场景下，使用 Signal 有助于避免代码中不必要的耦合，提高可维护性；并且，部分工程化实践，比如针对特定逻辑进行的测试，需要借助 Signal 的帮助才能完成（后面我们会看到一个具体的例子）。因此，本文将帮助你了解什么是 Signal，它的原理、使用方法，以及它在 Flask 中有哪些实际应用。

从技术上讲，Signal 是设计模式中观察者（Observer），也叫做侦听器（Listener）的一种实现。对于熟悉设计模式的同学，看到这个术语应该就会大致了解它的含义了。通过 Signal，开发者可以定义一系列预期会发生的事件，而程序中其他部分可以订阅（subscribe）自己感兴趣的事件。这样作的主要益处在于，当以后需要添加新的行为时，可以只增加新的订阅者，而发布事件的部分保持不变，从而无需担心新增的代码会破坏原有的逻辑。这也是开闭原则的一种体现。

Flask 主要通过第三方库 Blinker 来支持 Signal。因此，在介绍 Flask Signal 之前，我们需要简单了解一下 Blinker。

Blinker

Blinker 实现了一套简单而灵巧的事件发布/订阅机制，它的核心概念就是 Signal。要使用 Blinker，我们像常规 Python 库一样安装它：

pip install blinker

代码示例

为了便于理解，我用代码模拟一个大家都很熟悉的事件（按钮点击）作为案例：

from blinker import signal


class Button:
    def __init__(self, label):
        self.label = label
        self.clicked = signal('clicked')

    def click(self, pos):
        """触发事件"""
        self.clicked.send(self, pos=pos)

    def __str__(self):
        return 'Button({})'.format(self.label)


def on_click(sender, pos):
    """事件响应函数"""
    print('on_click, sender: {}, pos: {}'.format(sender, pos))


btn = Button('btn')
# 订阅事件
btn.clicked.connect(on_click)
btn.click((100, 200))

上述代码应该是比较容易理解的，但还是有些细节值得说明。通常来讲，处理事件分为三个步骤：

1. 定义要触发的事件；
2. 定义响应事件的方法(处理器)；
3. 将事件和响应方法关联起来（订阅）。

下面我们依次加以说明。

定义事件

定义事件需要调用 blinker.signal(name) 方法，并传递事件名称作为参数。很简单，对吧？有趣的一点是，signal 的实现是单例的，你不能重复定义一个已命名的事件。我们可以自己验证这一点：

s1 = signal('signal')
s2 = signal('signal')
print(s1 is s2)

运行一下代码，你会发现输出为 True。可以想象，如果多个第三方类库试图定义同名的事件，那么很容易引起冲突。因此，官方推荐的做法是：为每个库使用一个单独的命名空间：

from blinker import Namespace

my_events = Namespace()
my_click = my_events.signal('clicked')

事件处理器

事件处理器既可以是普通函数，也可以是类的方法。按照 Blinker 的约定，处理器接受的第一个参数应该是触发事件的对象，后面可以跟随任意参数，但其顺序应该和事件传递过来的参数完全一致。在我们的代码示例中，除了调用者（sender）之外还接受一个额外参数，即按钮点击的位置（pos）。

在项目开发实践中，为了支持新的功能，有时候我们需要为事件添加额外的参数，但这样可能会导致与原有的处理器发生冲突。为避免此类问题，常见的办法是让事件处理器用具名参数接收所有新增参数：

def event_handler(sender, arg1, arg2, **kwargs):
    ...

有时候我们会在使用 Blinker 的第三方库中看到类似上面的代码。我们需要理解，添加 kwargs 主要是为了解决兼容性问题，它并不意味着处理器方法可以随心所欲地接受任何参数。

订阅事件

订阅事件的基本方法，如上所述，是调用 signal.connect() 方法。

有时候，我们可能会希望只接受特定来源的事件。Blinker 通过为 connect() 方法提供可选的 sender 参数来实现这一点。因此，类似下面代码：

signal1.connect(handler1)
signal2.connect(handler2, sender=my_obj)

handler1 会收到所有 signal1 触发的事件，但只有 sender=my_obj 的事件才会被 handler2 接收到。

针对特定来源处理事件也是一种常见的模式，比如在 Flask 中定义的核心事件，其 sender 通常都指向 current_app 或其所代理的对象。因此，Blinker 又提供了一种基于装饰器的语法来关联事件，即 connect_via：

@my_signal.connect_via(app)
def signal_handler():
    ...

Flask Signal

在 Flask 中内置了对 Signal 的支持，但同时它是一个可选功能。也就是说，即使完全不了解 Signal，也不妨碍你成功地运行一个 Flask 程序。

这样讲似乎还有点抽象，其实说穿了也很简单。Flask 会自动检查 Blinker 库是否存在，如果是，就导入并使用它；否则，通过 Flask 自己编写的一小块“垫片”代码，允许上层以透明的方式使用 signal，不用管它究竟是 blinker signal，还是 Flask 内部实现的。

如果你去翻阅源码的话，可以在 flask/signals.py 文件中看到它是如何实现的：

try:
    from blinker import Namespace
    signals_available = True
except ImportError:
    signals_available = False
    class Namespace(): ...
_signals = Namespace()    

Flask 内置事件

我们已经介绍了 Flask 是如何支持 Signal 的。在其源码实现的 flask/signals.py 文件最后，我们还会看到 Flask 自身定义的一些核心信号。大家应该还记得，信号的首个参数通常应该是一个名为 sender 的变量，表示信号发出的对象。对于 Flask 核心信号来说，sender 通常就是应用程序即 current_app。但实际上，更准确的用法是使用 current_app._get_current_object()。为什么要这么复杂呢？熟悉 Flask 的朋友可能知道，包括 App 在内的很多对象本质上是一个代理，为了支持多用户并发请求，它被设计为根据当前所在线程指向不同的具体实例。这也是 _get_current_object() 方法存在的意义。当然，如果你很确定事件的处理不需要涉及 Flask Context，那么简单地使用 current_app 也是可以的。

Flask 内置信号简单总结如下：

请注意，所有信号的第一个参数都是 sender，它们通常指向 Flask App 的绑定上下文。为了简洁起见，上表中并未包括它们。部分事件包含额外的参数，以便开发者获得关于触发信号的详细信息。举例来说，如果你想要侦听模板渲染的事件，那么一般的写法应该类似这样：

from flask import template_rendered

def handle_template_rendered(sender, template, context, **extra):
    # TODO: process signal
template_rendered.connect(handle_template_rendered)    

第三方扩展和应用中的信号

由于 Flask 内置了对 Signal 的支持，因此第三方扩展也可以使用信号。事实上，很多扩展确实使用了信号，比如我们熟悉的 Flask-SqlAlchemy，就定义了诸如 model_committed 的信号，以实现类似数据库触发器的机制。如果你对此感兴趣的话，请参阅本文后面列出的参考链接。稍后也会有一个例子说明如何使用 Flask-SqlAlchemy 中的信号。

假如你需要自己实现扩展，并作为软件包提供给其他人使用的话，那么你应该像 Flask 一样做到：即使没有安装 blinker，应用程序也能正常执行。因此，你需要使用 Flask 自身提供的 signal，而不是直接使用 blinker。典型的代码可能类似这样：

from flask.signals import Namespace
ext_events = Namespace()
ext_signal = ext_events.signal('my-signal')

当然，如果你是在 Web 应用中使用信号的话，就可以安装并直接使用 blinker，而无需担心它是否存在的问题。

应用

接下来，我们通过几个具体实例，体验 Signal 在实际中的应用。

案例1：测试模板渲染是否符合预期

对于常规的 Flask 应用，我们通常会在视图函数中调用 render_template 返回响应结果。但从测试的角度看，该方法存在一个问题：render_template 返回的是完整的 HTML，而从一大串 HTML 中检查是否包含我们预期的内容，势必要我们作很多手工解析的工作，复杂而且非常不合理。那么如何才能有效率地执行测试呢？我们回忆上面提到的 Flask 核心信号，其中包含 template_rendered，该信号包含的参数 template/context 恰好是我们关注的内容。因此，合理使用信号，我们能够更简单高效地完成单元测试工作。

为此，我们建立一个基本的 Flask 应用：

from flask import Flask, render_template


app = Flask(__name__)


@app.route('/')
def index():
    return render_template('index.html', name='guest')


if __name__ == '__main__':
    app.run()

然后编写单元测试：

from unittest import TestCase
from flask import template_rendered
from app import app


class ViewsTest(TestCase):
    def test_index(self):
        render_params = {'template': None, 'context': None}

        def on_template_rendered(sender, template, context):
            render_params.update({'template': template, 'context': context})
        try:
            template_rendered.connect(on_template_rendered)
            resp = app.test_client().get('/')
            template, context = render_params['template'], render_params['context']
            assert resp.status_code == 200
            assert template.name == 'index.html'
            assert context['name'] == 'guest'
        finally:
            template_rendered.disconnect(on_template_rendered)

为了保证每个测试完毕之后重置状态，我们在代码中使用了取消订阅的方法：disconnect。通过跟踪信号参数，我们获得了模板参数（template/context），并检查它们是否符合我们的预期。整个代码还是比较容易理解的。最重要的是，如果不通过信号，我们将无法直接检查这些参数，只能自己解析返回的 HTML，这是个烦人且容易出错的工作。

如果需要对不同的视图执行大量类似的测试，你或许可以考虑将上述代码封装成一个比较通用的函数。

案例2：Flask SqlAlchemy

Flask-SqlAlchemy 定义了两个信号：

• models_committed: 在数据变更提交时触发
• before_models_committed：在数据变更提交之前触发。

这两个信号的主要区别是，你可以在 before_models_committed 的时候修改字段数据，而在 models_committed 发生时，提交已经完成，再修改也没用了。

上述事件均接受两个参数：sender/changes。changes 是一个包含 tuple 的序列，每条 tuple 又有两项内容：要修改的实体（model）以及变更方式（operation，可能的值包括 insert/update/delete）。

为了说明其用法，我们假定一个简化的博客场景，其中每条博文包含内容和简短的汇总，而汇总不需要作者自己修改，而是从内容的前面一段文字自动生成。为实现该要求，我们首先定义模型：

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.Text(), nullable=True)
    summary = db.Column(db.String(128), nullable=True)

至于配置 SqlAlchemy 的部分这里就不再展示了，相信熟悉 Flask 的朋友都能自己完成。唯一需要说明的是，为了让信号正常工作，我们需要在配置中指定 SQLALCHEMY_TRACK_MODIFICATIONS = True。

def on_before_models_committed(sender, changes):
    for model, operation in changes:
        if isinstance(model, Post) and operation in ('insert', 'update'):
            if model.content and not model.summary:
                model.summary = model.content[:100]
before_models_committed.connect(on_before_models_committed)    

上述代码实现了简单的处理逻辑：如果没有指定汇总字段（summary），那么我们就从内容（content）提取前面部分的内容，自动生成汇总。由于该信号对所有实体都会触发，所以你应该仔细判断，保证当前的实体确实是我们预期的内容。

当然，我们这里提供的并不是针对该需求唯一的实现方式。有的朋友可能会想到，通过 SqlAlchemy 内置的 event 机制也可以实现类似的功能，此外，数据库触发器也是一种可以考虑的选择。

总结

本文介绍如下内容：

• Signal 的基础模块 blinker；
• Flask 对 Blinker Signal 的封装；
• Flask 的核心信号与扩展；
• 通过实例了解 Flask Signal 的应用场景和使用方法。

希望本文能够帮助大家更好地了解 Flask Signal，当然也希望借助该工具，能够想象出更多的应用场景，以充分发挥其作用，实现更加高内聚、低耦合的代码。

参考

• Blinker 官网
• Flask Signals
• Flask-SqlAlchemy Signals