说说 Flask 中的 Signal

版权声明:所有博客文章除特殊声明外均为原创,允许转载,但要求注明出处。

概述

信号(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 内置信号简单总结如下:

信号 参数 描述
template_rendered template, context 模板成功渲染之后触发
before_render_template template, context 模板渲染之前触发
request_started 请求对象已构造,但尚未开始处理
request_finished response 请求处理完毕,但应答尚未开始发送
request_tearing_down 整个请求处理完毕。即使出现异常,该信号也会触发
got_request_exception exception 请求处理过程出现异常,且常规异常处理尚未执行
appcontext_tearing_down 处理结束,应用上下文将被关闭
appcontext_pushed 应用上下文入栈,通常是因为新的请求到来,也可能由于开发者手工调用了 appcontext.push()。
appcontext_popped 应用上下文出栈
message_flashed message, category 开发者调用了 flash() 接口以添加闪现(flash)信息

请注意,所有信号的第一个参数都是 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
  • FlaskBlinker Signal 的封装;
  • Flask 的核心信号与扩展;
  • 通过实例了解 Flask Signal 的应用场景和使用方法。

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

参考