0.前言：
最近使用flask搭建博客实现登录时，碰到登录用户无法分配Permissions的问题，整整花了一个星期的时间，所以想记录一下。
1.Signal基本了解：
根据官方文档给出的解释：
What are signals? Signals help you decouple applications by sending notifications when actions occur elsewhere in the core framework or another Flask extensions. In short, signals allow certain senders to notify subscribers that something happened.
Signal实际上是用于解耦系统中实现某种业务逻辑和行为。所谓的解耦，就是某些行为被触发时，自动发送定义好的一种信号，与这个信号绑定的一些业务逻辑或行为，接收到这个信号后，会自动执行各自相应的业务逻辑。信号和一些业务逻辑或行为绑定好了之后，只需要发送一个信号即可，所有与该行为有关的业务逻辑或行为都会自动触发，从而实现了解耦。信号发送无需了解接收信号的订阅者是谁，这就是观察者模式的一种实现方式。
2.Signal实现原理：
from flask.signals import Namespace
from flask import current_app
signals = Namespace()
mysignal = signals.signal('save-models') #signal的创建
def save_models:
...
mysignal.send(current_app._get_current_object_) #信号发送
Signal的订阅者:
一些订阅了指定signal也就是上文中的mysignal的函数，当调用send()函数发送signal时，会自动触发这些订阅者的调用，以完成某些功能。将一个普通函数变为signal的订阅函数非常简单，只要加一个decorator即可，仍以上面signal为例，定义一个订阅者方法如下：
@mysignal.connect_via(app)
# mysignal.connect也可以
def on_model_saved():
# do something ...
3.踩坑：
上面代码看起来是不是很简单？接下来就开始讲之前碰到的坑：
其实在Flask-Principal应用场景解析中有讲过，flask在用户登录时实现身份改变信号逻辑中，通过identity_changed和identity_loaded这两个不同的signal来处理的，当用户登录之前，我们都会通过发送一个信号说明用户身份已经改变，当相应的逻辑处理接收到信号之后，能后改变其身份信息以及权限信息，login_view.py:
from flask import Flask, current_app, request, session
from flask_login import LoginManager, login_user, logout_user, login_required, current_user
from flask_principal import identity_changed
@app.route('/login', methods=['GET', 'POST'])
def login():
form = LoginForm()
if form.validate_on_submit():
user = datastore.find_user(email=form.email.data)
if form.password.data == user.password:
# Flask-Login的login_user方法将登录用户信息保存于session中
login_user(user)
# 发送信号
identity_changed.send(current_app._get_current_object(),
identity=Identity(user.id))
return redirect(request.args.get('next') or '/')
return render_template('login.html', form=form)
Signal订阅者on_identity_change：
from flask_principal import identity_loaded, RoleNeed, UserNeed, Permission
from flask_login import current_user
admin_need = RoleNeed('admin')
admin_Permission = Permission(admin_need)
@identity_loaded.connect
def on_identity_change(sender, identity):
identity.user = current_user
if hasattr(current_user, 'username'):
identity.provides.add(UserNeed(current_user.username))
if hasattr(current_user, 'role'):
identity.provides.add(RoleNeed(current_user.role))
if hasattr(current_user, 'admin')
identity.provides.add(admin_need)
运行后发现{'identity': <Identity id="admin" auth_type="None" provides=set()>}provides为空
然后在debug跟踪发现，receivers为空的dict，不是说过只要是有定义的signal发送信号，再绑定对应的订阅者就会自动触发执行该函数吗？
接下来，就开始看源代码，看看执行流程：
1.在工厂函数初始化创建app时，就有init_app函数进行初始化,identity_changed.connect(self._on_identity_changed, app)来进行连接：
from flask import Flask
from flask_login import LoginManager
from flask_principal import Principal
login_manager = LoginManager()
login_manager.session_protection = 'basic'
login_manager.login_view = 'xxx.login'
Principals = Principal()
def create_app(config_name):
app = Flask(__name__)
...
login_manager.init_app(app)
Principals.init_app(app)//初始化app
return app
app = create_app()
对应源码：
class Principal(object):
...
...
def init_app(self, app):
if hasattr(app, 'static_url_path'):
self._static_path = app.static_url_path
else:
self._static_path = app.static_path
app.before_request(self._on_before_request)
identity_changed.connect(self._on_identity_changed, app)
if self.use_sessions:
self.identity_loader(session_identity_loader)
self.identity_saver(session_identity_saver)
2.接着通过_on_identity_changed来进行调用，接着调用set_identity，源码：
class Principal(object):
...
...
def _on_identity_changed(self, app, identity):
if self._is_static_route():
return
self.set_identity(identity)
3.接着调用_set_thread_identity来执行identity_loaded.send()发送identity_loaded信号,源码：
class Principal(object):
...
...
def set_identity(self, identity):
"""Set the current identity.
:param identity: The identity to set
"""
self._set_thread_identity(identity)
for saver in self.identity_savers:
saver(identity)
def _set_thread_identity(self, identity):
g.identity = identity
identity_loaded.send(current_app._get_current_object(),
identity=identity)
4.接下来，我们看send函数的源码：
def send(self, *sender, **kwargs):
"""Emit this signal on behalf of *sender*, passing on \*\*kwargs.
Returns a list of 2-tuples, pairing receivers with their return
value. The ordering of receiver notification is undefined.
:param \*sender: Any object or ``None``. If omitted, synonymous
with ``None``. Only accepts one positional argument.
:param \*\*kwargs: Data to be sent to receivers.
"""
# Using '*sender' rather than 'sender=None' allows 'sender' to be
# used as a keyword argument- i.e. it's an invisible name in the
# function signature.
if len(sender) == 0:
sender = None
elif len(sender) > 1:
raise TypeError('send() accepts only one positional argument, '
'%s given' % len(sender))
else:
sender = sender[0]
if not self.receivers:
return []
else:
return [(receiver, receiver(sender, **kwargs))
就在程序执行到这一步中，发现self.receivers为dict{ },为什么找不到接收信号的订阅者函数？其实我们回过头来看signal基本实现机制就知道，当定义一个signal之后发送信号，所以与这个信号有关的业务逻辑或者行为都会自动触发，自动触发的条件就是与这个信号有关，也就是说，我们要用引用到这个条件才行。所以在login_view.py中，引用订阅者函数对应的模块：
from flask import Flask, current_app, request, session
from flask_login import LoginManager, login_user, logout_user, login_required, current_user
from flask_principal import identity_changed
from .permissions import admin_Permission # 引用Permissions模块
@app.route('/login', methods=['GET', 'POST'])
def login():
form = LoginForm()
if form.validate_on_submit():
user = datastore.find_user(email=form.email.data)
if form.password.data == user.password:
# Flask-Login的login_user方法将登录用户信息保存于session中
login_user(user)
# 发送信号
identity_changed.send(current_app._get_current_object(),
identity=Identity(user.id))
return redirect(request.args.get('next') or '/')
return render_template('login.html', form=form)
总结：
1.要熟悉实现某种功能使用到的某个模块的实现原理和流程
2.善于研究源码，熟悉flask一些机制
3.记得利用stack overflow等一些平台解决问题。
参考：
Gevin大神的文章：https://blog.igevin.info/posts/flask-signal-get-started/
stackoverflow：https://stackoverflow.com/questions/7050137/flask-principal-tutorial-auth-authr/9781669#9781669
官方文档：http://flask.pocoo.org/docs/1.0/signals/
blinker文档：https://pythonhosted.org/blinker/
http://www.bjhee.com/flask-ad2.html
