Flask信号机制
Flask各模块之间通过信号signal进行通信,信号的存在使得各模块之间不需要进行互相调用,也就是所谓的解耦合。信号对于我们进行单元测试也是特别的重要。
信号和一些内置的装饰器比较类似,如request_started和before_request,虽然类似,但是他们工作方式完全不同。before_request是按顺序执行,并返回一个响应类;然而request_started信号仅仅是一个信号,只负责通知,而不修改任何的数据,并且是无序的。
一个信号signal存在一个发送者sender,一个订阅者subscriber。发送者不需要知道谁是订阅者,订阅者也可以自由选择订阅哪些消息。那么就分别说说这两个信号对象。
先来个简单的例子:
from blinker import Namespace
my_signal = Namespace()
test_signal = my_signal.signal('test-signal')
def add(num):
print num + 1
def mux(num):
print num*2
#signal reciever
test_signal.connect(add,sende=0)
test_signal.connect(mux,sender=2)
#signal sender
for num in range(0,3):
test_signal.send(num)
信号的发送者通过send方法实现;信号的接收者通过connect,可订阅对应函数,但都有一个sender限制,只有sender是对应值的时候才被接收。
看到了吧,上面的add和num函数,我们并没有直接显示调用他们,而是通过信号就可以调用了,通过信号test_signal作为桥梁,就实现了解耦合的作用。
发送信号
上面也提到了,发送信号的发送者是一个信号类,Flask采用了blinker的信号类。创建信号很简单:
from blinker import Namespace
my_signals = Namespace()
model_saved = my_signals.signal('model-saved')
要发送信号的话就使用信号类的send方法,send方法第一个参数是发送者,后面可以有多个参数,但是必须是key=value的形式。原型:send(self,*sender, kwargs)。其中,sender是发送者,kwargs是可选参数,是传递订阅者的数据。
如:模板成功渲染后,就要发送信号,可选参数就是模板和上下文,template_rendered.send(app, template=template, context=context)。源代码:
def render_template(template_name_or_list, context):
"""Renders a template from the template folder with the given
context.
"""
ctx = _app_ctx_stack.top
ctx.app.update_template_context(context)
return _render(ctx.app.jinja_env.get_or_select_template(template_name_or_list),
context, ctx.app)
def _render(template, context, app): """Renders the template and fires the signal"""
before_render_template.send(app, template=template, context=context)
rv = template.render(context)
template_rendered.send(app, template=template, context=context)
return rv
订阅信号
使用信号的connect()方法可以订阅该信号。该方法的原型是:connect(receiver,sender=ANY,weak=True)
- receiver:信号发出时将调用的函数。connect将回调函数和发送者联系了起来。
- sender:该参数是可选的,用于确定信号的发送者。所有核心Flask信号的发送者是应用本身。因此当订阅信号时请指定发送者,除非你真的想要收听应用的所有信号。当你在开发一个扩展时,尤其要注意这点
- weak:如果是True,那么信号与接收者之间是弱引用。当接收者超出范围或被垃圾回收时,信号与接收者自动断开连接。
- 也可以使用disconnect()方法来退订信号。该方法原型为:disconnect(receiver,sender=ANY) 其中:receiver为先前connect()函数的第一个参数receiver。sender为信号的发送者。
Blinker使得我们可以通过装饰器实现订阅:
from flask import template_renderd
@template_rendered.connect_via(app)
def when_template_rendered(sender,template,context,**extra):
print 'Template %s is rendered with %s' %(template.name,context)
当templater_rendered发送信号时,上述函数会被调用,会输出Template is rendered with ****。
Flask核心信号
在Flask中订阅了几种核心信号:
-
flask.template_rendered 这个信号发送于一个模板被渲染成功后。信号传递的template是模板的实例,context是环境对象是一个字典。
-
flask.request_started 这个信号发送于请求开始之前,且请求环境设置完成之后。因为请求环境已经绑定, 所以订阅者可以用标准的全局代理,如 request 来操作请求。
订阅示例:def log_request(sender, **extra): sender.logger.debug('Request context is set up') from flask import request_started request_started.connect(log_request, app)
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是异常对象。
订阅示例:
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 这个信号发送于请求崩溃的时候,不管是否引发异常。目前,侦听此信号的函数在一般 崩溃处理器后调用,但是没有什么东西可用。 订阅示例:
def close_db_connection(sender, **extra): session.close()from flask import appcontext_tearing_down request_tearing_down.connect(close_db_connection, app)
-
flask.appcontext_tearing_down 当应用环境崩溃时发送这个信号。这个信号总是会发送,甚至是因为一个异常引发的 崩溃。侦听这个信号的函数会在常规崩溃处理器后被调用,但是你无法回馈这个信号。 订阅示例:
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对象中现存的资源。
用法示例:
from contextlib import contextmanagerfrom
flask import appcontext_pushed
@contextmanagerdef user_set(app, user):
def handler(sender, **kwargs):
g.user = user
with appcontext_pushed.connected_to(handler, app):
yield
在测试代码中这样写:
def test_user_me(self):
with user_set(app, 'john'):
c = app.test_client()
resp = c.get('/users/me')
assert resp.data == 'username=john'
-
appcontext_popped 当一个应用的环境被弹出时,应用会发送这个信号。这个信号通常写成appcontext_tearing_down 信号。
-
flask.message_flashed
当应用闪现一个消息时会发出这个信号。message 参数是消息内容,
category参数是消息类别。 订阅示例:
recorded = []def record(sender, message, category, **extra):
recorded.append((message, category))
from flask import message_flashed
message_flashed.connect(record, app)
微信分享/微信扫码阅读