0.6新版功能

从Flask 0.6开始，Flask继承了信号支持。这个支持由blinker库提供的，并且当它不可用时会优雅地退回。

什么是信号？信号通过发送发生在和新框架的其他地方或Flask扩展的动作时通知来帮助你解耦应用。简而言之，信号允许特定的发送端通知订阅者发生了什么。

Flask提供了几个信号，其他的扩展可能会提供更多。另外，请注意信号倾向于通知订阅者，而不应该鼓励订阅者修改数据。你会注意到，信号似乎和一些内置的装饰器做同样的事情。然后它们工作的方式是有差异的。譬如核心的before_request()处理程序以特定的顺序执行，并且可以在返回响应之前放弃请求。相比之下，所有信号的处理器执行的顺序没有定义，并且不修改任何数据、

信号之于其他处理器最大的优势是你可以在一秒钟的不同的时段上安全订阅。譬如这些临时的订阅对单元测试很有用。比如你想要知道哪个模板被作为请求的一部分渲染：信号允许你完全地了解这些。

订阅信号

你可以使用信号的connect()方法来订阅信号。该函数第一个参数是信号发出时要调用的函数，第二个参数是可选的，用于确定信号的发送端。退订一个信号，可以使用disconnect()方法。

对于所有的核心Flask信号，发送端都是发出信号的应用。当你订阅一个信号，情趣噩耗也提供一个发送端，除非你确实详见听全部应用的信号。这在你开发一个扩展的时候尤其正确。

比如这里有一个用于在单元测试中找出哪个模板被渲染和传入模板的变量的助手上下文管理器：

1
2
3
4
5
6
7
8
9
10
11
12
13
from flask import template_rendered
from contextlib import contextmanager

@contextmanager
def captured_temolates(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)

这可以很容易地与一个测试客户端配对：

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

确保订阅使用了一个额外的**extra参数，这样当Flask对象引入新参数时你的调用不会失败。

代码中，从with块的应用app中流出的渲染的所有模板现在会被记录到templates变量。无论何时模板被渲染，模板对象和上下文中都会被添加到它里面。

此外，也有一个方便的助手方法(connected_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, ttemplates, **extra):
    ...
    template, context = templates[0]

注：connected_to()方法出现于Blinker 1.1

创建信号

如果你想要在自己的应用中使用信号，你可以直接使用blinker库。最常见的用法是自定义的Namespace中命名信号。这也是大多数时候推荐的做法：

1
2
from blinker import Namespace
my_signals = Namespace()

现在你可以这样创建新的信号：

1
model_saved = my_signals.signal("model-saved")

这里使用唯一的信号名，简化调试。可以用name属性来访问信号名。

给扩展开发者

如果你在编写一个Flask扩展并且你想优雅地在没有blinker安装时退化，你可以用flask.signals.Namespace这么做。

发送信号

如果你想要发出信号，调用send()方法可以做到。它接受发送端作为第一个参数，和一些推送到信号订阅者的可选关键字参数：

1
2
3
4
5
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_started和request_finished一贯可用，所以你可以新人flask.g和其他需要的东西。注意发送信号和request_tearing_down信号中描述的限制。

基于装饰器的信号订阅

你可以在Blinker 1.1中容易地用新的connect_via()装饰器订阅信号：

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))

核心信号

下列是Flask中存在的信号：

flask.template_rendered

当模板成功渲染的时候，这个信号就会发出。这个信号与模板实例template和上下文的字典(名为context)一起调用。

订阅示例：

1
2
3
4
5
6
def log_template_renders(sender, template, context, **extra):
    sender.logger.debug('Rendering template "%s" with context %s', template.name or 'string template', context)
    
from flask import template_rendered

template_rendered.connect(log_template_renders, app)

flask.request_started

这个信号在此处建立请求上下文之外的任何请求处理开始前发送。因为请求上下文已经被约束，订阅者可以用request之类的标准全局代理访问请求。

订阅示例：

1
2
3
4
5
6
def log_request(sender, **extra):
    sender.logger.debug('Request context is set up')
    
from flask import request_started

request_started.connect(log_request, app)

flask.request_finished

这个信号恰好在请求发送给客户端之前发送。它传递名为response的响应。

订阅示例：

1
2
3
4
5
def log_response(sender, response, **extra):
    sender.logger.debug('Request context is about to close down. Response: %s', response)
    
from flask import request_finished
request_finished.connect(log_response, app)

flask.got_request_exception

这个信号在请求处理中抛出异常时发送。它在标准一场处理生效之前，甚至是在没有异常处理的情况下发送。异常本身会通过exception传递订阅函数。

订阅示例：

1
2
3
4
5
def log_exception(sender, exception, **extra):
    sender.logger.debug('Got exception during processing: %s', exception)
    
from flask import got_request_exception
got_request_exception.connect(log_exception, app)

flask.request_tearing_down

这个信号在请求销毁时发送。他总是被调用，即使发生异常。当前监听这个信号的函数会在常规销毁处理后被调用，但这不是你可以依赖的。

订阅示例：

1
2
3
4
5
def close_db_connection(sender, **extra):
    session.close()
    
from flask import request_tearing_down
request_tearing_down.connect(close_db_connection, app)

从Flask 0.9，如果有异常的话他会被传递一个exc关键字餐朱引用导致销毁的异常。

flask.appcontext_tearing_down

这个消息在应用上下文销毁时发送。它总是被调用，即使发生异常。当前建英这个信号的函数会在常规销毁处理后被调用，但这不是你可以依赖的。

订阅示例：

1
2
3
4
5
def close_db_connection(sender, **extra):
    session.close()
    
from flask import request_tearing_down
appcontext_tearing_down.connect(close_db_connection, app)

如果有一场他会被传递一个exc关键字参数引用导致销毁的异常。

flask.appcontext_pushed

这个信号在应用上下文压入栈时发送。发送者是应用对象。这通常在单元测试中为了暂时地钩住信息比较有用。例如这个可以用来提前在g对象上设置一些资源。

用法示例：

1
2
3
4
5
6
7
8
9
from contextlib import contextmanager
from flask impo

@contextmanager
def user_set(app, user):
    def handler(sender, **kwargs):
        g.user = user
    with appcontext_pushed.connected_to(handler, app):
        yield

测试代码：

1
2
3
4
5
def test_user_me(self):
    with user_set(app, 'john'):
        c = app.test_client()
        resp = c.get('/users/me')
        assert resp.data == 'username=john'

0.10 新版功能

flask.appcontext_popped

这个信号在应用上下文弹出栈时发送。发送这是应用对象。这通常在appcontext_tearing_down信号发送后发送。

0.10 新版功能

flask.message_flashed

这个信号在应用对象闪现一个消息时发送。消息作为message命名参数发送，分别则是categpry参数。

订阅示例：

1
2
3
4
5
6
recorded = []
def record(sender, message, category, **extra):
    recorded.append((message, category))
    
from flask import message_flashed
message_flashed.connect(record, app)

0.10 新版功能

文章借鉴Flask中文手册