Flask Signals 详解

Flask Signals和操作系统的signals系统很类似，都是通过信号（也可以说是事件 event）来通知已经注册的回调函数，让回调函数自动开始执行。Flask定义了自己 的一套核心signals和对应的functions(用于发起消息，注册回调函数)，我们需要 定义自己的回调函数，然后注册到对应的signal，这样就可以在收到该信号的时候 自动执行我们定义的回调函数。

Flask Signals简介

Flask Signals和操作系统的signals系统很类似，都是通过信号（也可以说是事件event）来通知已经注册的回调函数，让回调函数自动开始执行。Flask定义了自己的一套核心signals和对应的functions(用于发起消息，注册回调函数)，我们需要定义自己的回调函数，然后注册到对应的signal，这样就可以在收到该信号的时候自动执行我们定义的回调函数。

什么情况下需要使用Signals?

当我们需要使用观察者模式来解耦模块之间的信息传递的时候，Signals系统就可以帮助我们轻松达到目的。观察者模式如下图(图片来自voidcn) 观察者模式

与Hook函数的区别

试想，当我们需要监听某个事件，当它发生的时候，需要执行一系列functions，来实现诸如log记录等功能时，我们就可以使用Signals系统来实现，但是这里有一个疑问就是这个功能通过hook函数似乎也可以实现，比如通过before_request decorator实现记录日志的功能和使用request_started来记录日志就非常相似， 如下代码所示： 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#!/usr/bin/env python
# -*- coding: utf-8 -*-

from flask import Flask, request, request_started
app = Flask(__name__)

@app.before_request
def print_url_in_hook():
    print "in hook, url: %s" % request.url

@app.route("/")
def hello():
    return "Hello, World!"

def print_url_in_signal_subscriber(sender, **extra):
    print "in signal subscriber, url: %s" % request.url

if __name__ == "__main__":
    request_started.connect(print_url_in_signal_subscriber, app)
    app.run()

当收到http请求后，打印如下： 

1
2
3
in signal subscriber, url: http://localhost:5000/
in hook, url: http://localhost:5000/
127.0.0.1 - - [05/Oct/2017 16:57:20] "GET / HTTP/1.1" 200 -

那么到底什么情况下使用signal，什么情况下使用hook函数呢？我们来看下它们的主要区别：

  1. signal的callback函数是无顺序的，而hook函数的执行是按照定义的顺序执行的。（这一点虽然是官网提出的区别，但是实际测试发现signal执行实际是按照注册的顺序执行的，即先通过connect进行注册的回调函数会先被执行）
  2. signal无法直接abort这个request请求，相比较在hook函数中可以直接abort request，即直接返回response给客户端，而无需再执行后续的操作。
  3. signal可以通过参数携带数据，而hook函数通常不会携带额外的参数

与RabbitMQ等消息中间件的区别

Rabbitmq与signals都支持观察者模式，但是它们的区别也是很明显的：

  1. Rabbitmq之类的消息中间件更加重量级，提供更多功能，如分布式部署，消息存储备份等功能，而signal系统显然更加轻量级，只提供简单的消息分发功能
  2. Rabbitmq之类的消息中间件可以在不同的系统间传递消息，从而使得不同的功能模块可以使用不同的语言进行开发，而signal系统显然仅限于Flask系统中使用

显然，signal系统使用局限性更大，但也更加轻量级，在只是简单的进行消息分发的系统中，使用signal更加简单方便

怎么使用Signals?

Flask提供的signal机制优先使用blinker提供的库，但当blinker没有安装的时候，Flask也可以回退到使用自己的库。但是鉴于官网推荐使用blinker，所以我们最好还是安装blinker。

使用blinker

安装blinker 

1
pip install blinker

测试Flask signal是否使用blinker 

1
2
3
4
In [1]: from flask import signals

In [2]: signals.signals_available
Out[2]: True

当signals.signals_available返回True时，说明使用的是Blinker库

使用Flask Built-in signals

Flask内置有多个signals可以直接使用，这些signals会自动emit(发射)，我们只需要定义自己的回调函数，然后通过connect方式来subscribe我们定义的函数到对应的signal即可监听该signal

下表展示了Flask内置的Signals，详细请参考Flask built-in signals:

Signals 说明
template_rendered 当template被成功渲染之后会触发
before_render_template 当template被渲染之前会触发
request_started 当request context建立好之后，并在request被处理之前
request_finished 当发送response给客户端之后被触发
got_request_exception 当request处理过程中发生异常时，该signal会被触发，它甚至早于程序中的异常处理
request_tearing_down 当request tear down的时候触发，无论何种情况该signal都会被触发，即使发生异常
appcontext_tearing_down 当应用的context tear down的时候触发
appcontext_pushed 当应用的context被push时触发
appcontext_popped 当应用的context被pop时触发
message_flashed 当应用发送flash message时触发

之前的例子我们已经看到如何使用request_started signal了，这里需要说明两点：

  1. 在定义回调函数时，第一个参数必须是sender对象（即发送该signal的对象），第二个参数**extra用于接受额外的参数，也防止将来Flask在发送signal时添加新的参数。
  2. 使用connect注册回调函数时，第一个参数是回调函数，这个是必须的，第二参数是sender对象，是可选的，但最佳实践是要明确发送该signal的对象

另外，我们也可以临时性注册一个回调函数，这个尤其在进行单元测试时非常有用，因为我们不想在实际程序中添加测试相关的回调函数，因此需要一种机制在测试完成后，再取消注册该回调函数，有两种方式可以此种临时注册的机制：

  • 一种是通过contextmanagerdecorator和disconnect函数一起来实现，如下： 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    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))
        # 当使用with关键字进入with context时，自动注册record函数到template_rendered signal
        template_rendered.connect(record, app)
        try:
            yield recorded
        finally:
            # with context结束时会自动调用disconnect函数来解除注册
            template_rendered.disconnect(record, app)

使用时代码如下： 

1
2
3
4
5
6
7
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

  • 另外一种方式是使用connect_to函数 
    1
    2
    3
    4
    5
    6
    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)

使用时代码如下： 

1
2
3
4
templates = []
with captured_templates(app, templates, **extra):
    ...
    template, context = templates[0]

自定义signals的使用

自定义signal

当我们需要自定义signal时，我们可以直接使用blinker库

  1. 首先定义一个namespace 

    1
    2
    from blinker import Namespace
    my_signals = Namespace()

  2. 使用我们自定义的namespace定义自己的signal 

    1
    upload_image_finished = my_signals.signal('upload_image_finished')

至此，我们就定义了一个signal，名为upload_image_finished

发射自定义signal

1
2
3
4
5
6
7
from flask import current_app

def upload_image(image_path, upload_url):
    # upload image code
    ...
    # after upload image
    upload_image_finished.send(current_app._get_current_object())

  • 当在类的method中使用send函数发射signal时，我们可以选择该类的对象作为sender对象，因此直接使用self作为参数，但是当我们不是在类的method当中，或者我们想让应用对象作为sender，那么我们就需使用如上代码所示的current_app._get_current_object()来获取应用对象
  • 使用sender时，第一个参数是sender对象，是必选的。其余实际我们还可以传递更多参数(记得我们的callback函数使用了**extra), 这样的话我们实际就拥有了传递更多数据的能力。

注册回调函数的简化写法

从文章的第一个示例可以看出我们需要通过调用connect函数来对回调函数进行注册, 其实还有一个简化的写法可以把回调函数的定义和注册过程结合在一起，如下： 

1
2
3
4
5
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)

通过connect_via装饰器来简化回调函数定义和注册的过程

Reference