信号就是在框架核心功能或者一些 Flask 扩展发生动作时发送的通知 , 利用信号可以实现一部分的业务解藕。

在 Flask 中 , 信号功能由 Blinker 库提供 , 如果没有安装该库就无法使用信号功能 , 但是不会影响其他功能 , 因为如果没有该库 , Flask 将提供一个假的信号 , flask.signals.py 中 :

  1. signals_available = False
  2. try:
  3.     from blinker import Namespace
  4.     signals_available = True
  5. except ImportError:
  6.     class Namespace(object):
  7.         def signal(self, name, doc=None):
  8.             return _FakeSignal(name, doc)
  10.     class _FakeSignal(object):
  11.         """If blinker is unavailable, create a fake class with the same
  12.         interface that allows sending of signals but will fail with an
  13.         error on anything else.  Instead of doing anything on send, it
  14.         will just ignore the arguments and do nothing instead.
  15.         """
  17.         def __init__(self, name, doc=None):
  18.             self.name = name
  19.             self.__doc__ = doc
  20.         def _fail(self, *args, **kwargs):
  21.             raise RuntimeError('signalling support is unavailable '
  22.                                'because the blinker library is '
  23.                                'not installed.')
  24.         send = lambda *a, **kw: None
  25.         connect = disconnect = has_receivers_for = receivers_for = \
  26.             temporarily_connected_to = connected_to = _fail
  27.         del _fail
  29. # The namespace for code signals.  If you are not Flask code, do
  30. # not put signals in here.  Create your own namespace instead.
  31. _signals = Namespace()

需要安装 blinker

  1. $ pip install blinker

下面是一个 Blinker 的示例：

  1. from blinker import signal
  3. # 创建信号
  4. started = signal('test-started')
  6. def each(round):
  7.     print("Round {}!".format(round))
  9. def round_two(round):
  10.     print("Only {}".format(round))
  12. # 订阅信号,each为接收者
  13. started.connect(each)
  14. # round_two为接收者,sender为发送者
  15. # 表示只有发送者为2时才接收
  16. started.connect(round_two, sender=2)
  18. for round in range(1,4):
  19.     # 发送信号
  20.     started.send(round)

Flask 中有一些钩子 , 如 before_request 和 after_request , 这些钩子不需要 Blinker 库并且允许你改变请求对象 (request) 或者响应对象 (response) , 而信号和钩子做的事情很像 , 只不过信号并不对请求对象和响应对象做改变 , 仅承担记录和通知的工作

内置信号

在 flask.signals.py 中我们可以看到 , Flask 内置了 10 个信号 :

  1. # 模板渲染成功时发送
  2. template_rendered = _signals.signal('template-rendered')
  4. # 模板渲染前发送
  5. before_render_template = _signals.signal('before-render-template')
  7. # 建立请求上下文后,在请求处理开始前发送
  8. request_started = _signals.signal('request-started')
  10. # 在响应发送给客户端之前发送
  11. request_finished = _signals.signal('request-finished')
  13. # 请求销毁时发送,无论请求成败都会发送
  14. request_tearing_down = _signals.signal('request-tearing-down')
  16. # 请求处理抛出异常时发送
  17. got_request_exception = _signals.signal('got-request-exception')
  19. # 应用上下文销毁时发送
  20. appcontext_tearing_down = _signals.signal('appcontext-tearing-down')
  22. # 应用上下文进栈中时发送
  23. appcontext_pushed = _signals.signal('appcontext-pushed')
  25. # 应用上下文出栈时发送
  26. appcontext_popped = _signals.signal('appcontext-popped')
  28. # 调用flask在其中添加数据时发送
  29. message_flashed = _signals.signal('message-flashed')

创建信号

我们以 request_started 为例来看看其内部实现 :

  1. from blinker import Namespace
  3. _signals = Namespace()
  5. # 调用Namespace对象的signal方法
  6. # 完成信号对象的创建,并使其成为全局引用
  7. request_started = _signals.signal('request-started')

Namespace.signal() 如下 :

  1. class Namespace(dict):
  2.     """A mapping of signal names to signals."""
  4.     def signal(self, name, doc=None):
  5.         """
  6.         返回NamedSignal对象
  7.         """
  8.         try:
  9.             return self[name]
  10.         except KeyError:
  11.             # Namespace为内置对象dict的派生类,
  12.             # 设置并返回值,
  13.             # self.request-started = NameSignal('request-started')
  14.             return self.setdefault(name, NamedSignal(name, doc))

订阅信号

如果我们要使用内置信号 , 那么首先我们需要订阅信号 , 也就是使用 Signal.connect() 方法

  1. from flask import Flask, request_started
  3. app = Flask(__name__)
  5. # log_reqeust函数为接收方,app为发送方
  6. # 对于接收函数的参数,第一个位置不可缺省,
  7. # 因为在send调用该函数时,内部传入了一个sender实参
  8. def log_request(sender, **extra):
  9.     print('Before the request comes ...')
  11. # 订阅信号
  12. request_started.connect(log_request, app)
  15. @app.route('/index')
  16. def index():
  17.     return 'index page'
  20. if __name__ == '__main__':
  21.     app.run()

connect() 源码如下 :

  1. def connect(self, receiver, sender=ANY, weak=True):
  2.     """
  3.     Connect *receiver* to signal events sent by *sender*.
  4.     receiver:为一个可调用对象
  5.     """
  6.     receiver_id = hashable_identity(receiver)
  7.     if weak:
  8.         # <weakref at 0x000002129F72EE28; to 'function' at 0x000002129CBE7F28 (log_request)>
  9.         # receiver将在send时被调用,self._cleanup_receiver
  10.         receiver_ref = reference(receiver, self._cleanup_receiver)
  11.         receiver_ref.receiver_id = receiver_id
  12.     else:
  13.         receiver_ref = receiver
  14.     if sender is ANY:
  15.         sender_id = ANY_ID
  16.     else:
  17.         sender_id = hashable_identity(sender)
  19.     self.receivers.setdefault(receiver_id, receiver_ref)
  20.     # self._by_sender与self._by_receiver为两个默认字典,其value默认为set
  21.     # {sender_id:{receiver_id,}}
  22.     self._by_sender[sender_id].add(receiver_id)
  23.     # {receiver_id:{sender_id,}}
  24.     self._by_receiver[receiver_id].add(sender_id)
  25.     del receiver_ref
  27.     # 此时self._weak_senders为空,所以以下不会执行
  28.     if sender is not ANY and sender_id not in self._weak_senders:
  29.         # wire together a cleanup for weakref-able senders
  30.         try:
  31.             sender_ref = reference(sender, self._cleanup_sender)
  32.             sender_ref.sender_id = sender_id
  33.         except TypeError:
  34.             pass
  35.         else:
  36.             self._weak_senders.setdefault(sender_id, sender_ref)
  37.             del sender_ref
  39.     # 此处条件不成立,也不会执行
  40.     if ('receiver_connected' in self.__dict__ and
  41.         self.receiver_connected.receivers):
  42.         try:
  43.             self.receiver_connected.send(self,
  44.                                          receiver=receiver,
  45.                                          sender=sender,
  46.                                          weak=weak)
  47.         except:
  48.             self.disconnect(receiver, sender)
  49.             raise
  50.     # receiver_connected为空
  51.     if receiver_connected.receivers and self is not receiver_connected:
  52.         try:
  53.             receiver_connected.send(self,
  54.                                     receiver_arg=receiver,
  55.                                     sender_arg=sender,
  56.                                     weak_arg=weak)
  57.         except:
  58.             self.disconnect(receiver, sender)
  59.             raise
  60.     return receiver

发送信号

信号的发送是通过 Signal.send() 来完成的 , 而这一步早已经被定义在 Flask 对象中了 , 如下 :

  1. def full_dispatch_request(self):
  2.     self.try_trigger_before_first_request_functions()
  3.     try:
  4.         # 请求处理前发送信号
  5.         request_started.send(self)
  6.         rv = self.preprocess_request()
  7.         if rv is None:
  8.             # 分派请求
  9.             rv = self.dispatch_request()
  10.     except Exception as e:
  11.         rv = self.handle_user_exception(e)
  12.     return self.finalize_request(rv)

Signal.send() 如下 :

  1. def send(self, *sender, **kwargs):
  2.     # Using '*sender' rather than 'sender=None' allows 'sender' to be
  3.     # used as a keyword argument- i.e. it's an invisible name in the
  4.     # function signature.
  5.     if len(sender) == 0:
  6.         sender = None
  7.     elif len(sender) > 1:
  8.         raise TypeError('send() accepts only one positional argument, '
  9.                         '%s given' % len(sender))
  10.     else:
  11.         # 取*sender元组中的第一个元素,即self (app) 
  12.         sender = sender[0]
  13.     if not self.receivers:
  14.         return []
  15.     else:
  16.         # 返回并完成调用
  17.         return [(receiver, receiver(sender, **kwargs))
  18.                 for receiver in self.receivers_for(sender)]

Signal.receivers_for() 如下 :

  1. def receivers_for(self, sender):
  2.     # self.receivers在信号订阅时被设置
  3.     if self.receivers:
  4.         sender_id = hashable_identity(sender)
  5.         if sender_id in self._by_sender:
  6.             # 按照上面的例子我们使用的sender不是ANY,
  7.             # 所以self._by_sender[ANY_ID]为一个空集合,
  8.             # {sender_id:{receiver_id,}}
  9.             # self._by_sender[sender_id]为本例ids
  10.             ids = (self._by_sender[ANY_ID] |
  11.                    self._by_sender[sender_id])
  12.         else:
  13.             ids = self._by_sender[ANY_ID].copy()
  14.         for receiver_id in ids:
  15.             # 根据receiver_id获取weakref对象
  16.             receiver = self.receivers.get(receiver_id)
  17.             if receiver is None:
  18.                 continue
  19.             if isinstance(receiver, WeakTypes):
  20.                 # strong为订阅函数,即本例的log_reqeust
  21.                 # 这里你可能会疑惑,见下
  22.                 strong = receiver()
  23.                 if strong is None:
  24.                     # 释放信号
  25.                     self._disconnect(receiver_id, ANY_ID)
  26.                     continue
  27.                 receiver = strong
  28.             # 返回函数对象
  29.             yield receiver

在上面这段代码中 , 对于 strong = receiver() 我们知道 , WeakTypes = (ref, BoundMethodWeakref) , 而在这两个类型中 , ref 才是正主 ; 不用想我们也知道 , ref 也就是 ReferenceType 中必然有 call 方法 , 但是该方法仅仅一个 pass 摆在那里 , 而调用的返回值却返回了我们的订阅函数 , 这不正常
于是 , 在 ReferenceType 的上方我找到了说明 , Weak-reference support module
这个类型是一个弱引用类型 , 它是一个特殊的存在 , 当你对弱引用对象进行引用时 , 并不能保持该类对象的活动 , 只有通过调用引用判断 ; 如果该引用还存活着 , 那么将返回其引用对象 , 否则将会进行回调
大致过程如下 :

  1. # 依次调用代码
  2. receiver_ref = reference(receiver, self._cleanup_receiver)
  4. weak = callable_reference(object, callback)
  6. return annotatable_weakref(object, callback)
  8. class annotatable_weakref(ref):

弱引用对象没有属性或方法 , 如下有一个示例 :

  1. import weakref
  3. class Foo:
  4.     pass
  6. # 实例化Foo
  7. o = Foo()
  8. # 包装成弱引用对象
  9. r = weakref.ref(o)
  10. # 调用弱引用对象
  11. r_result = r()
  12. print(o is r_result)
  14. """
  15. 执行结果:
  16. True
  17. """

弱引用详见 : weakref

最后 , 对于其它信号的发送相关代码位置 , 我们可以通过导入信息来查看 , 导入信息如下 :

  1. # app.py (5个)
  2. from .signals import appcontext_tearing_down, got_request_exception, \
  3.     request_finished, request_started, request_tearing_down
  5. # ctx.py (2个)
  6. from .signals import appcontext_pushed, appcontext_popped
  8. # templating.py (2个)
  9. from .signals import template_rendered, before_render_template
  11. # helpers.py (1个)
  12. from .signals import message_flashed

自定义信号

我们可以直接使用 Blinker 创建信号 , 如下 , 定义一中对于上传大文件的信号 :

  1. from blinker import Namespace
  2. web_signals = Namespace()
  3. large_file_saved = web_signals.signal('large-file-saved')

装饰器方式

在 Signal 对象中还有一个 connect_via() 装饰器订阅信号

  1. def connect_via(self, sender, weak=False):
  2.     def decorator(fn):
  3.         self.connect(fn, sender, weak)
  4.         return fn
  5.     return decorator

以 flask.appcontext_tearing_down 为例：

  1. from flask import Flask, appcontext_tearing_down, session
  3. app = Flask(__name__)
  5. @appcontext_tearing_down.connect_via(app)
  6. def close_db_connection(sender, **extra):
  7.     print('Database connection closed ...')
  9. @app.route('/index')
  10. def index():
  11.     return 'index page'
  13. if __name__ == '__main__':
  14.     app.run()

另外在 Flask-Login 插件中还带了 6 种信号 , 可以基于其中的信号做一些额外工作