信号通过发送通知来帮助您解耦应用程序当操作发生在核心框架的其他地方或其他Flask扩展时,简言之，信号允许某些发送方通知订阅者发生了一些事情.
Flask 自带多个信号，其他扩展也可以提供更多。但是，记住这些信号是用来通知订阅方，但不应鼓励订阅方修改数据。你可能也注意到了，很多信号和内建的装饰器一样处理相同的事情。但是，不同的是他们的工作方式。例如：
before_request() 处理, 在特定的顺序下执行，可以终止请求在返回一个响应之前。相比较而言，很多信号处理方法执行时无序的，而且他们不修改任何数据。
信号的一大优势是，你可以快速的安全的订阅它。
1.订阅信号
为了订阅一个信号，你可以使用connect方法。方法的第一个参数是一个函数在信号发出之后调用。可选的第二个参数指定发送者。为了取消订阅一个信号，你可以使用disconnect方法。建议指定第二个参数
实例：
在单元测试中,
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))
template_rendered.connect(record, app)
try:
yield recorded
finally:
template_rendered.disconnect(record, app)
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
2.创建信号
如果想使用信号在我们的应用程序中，你可以直接使用blinker库。最常见的用法是使用Namespace.以下是推荐的用法：
from blinker import Namespace
my_signals = Namespace()
model_saved = my_signals.signal('model-saved') #创建一个新的信号
如果你想发送一个信号，你可以使用send方法。该方法接收发送方作为一个参数，可选的其他关键字参数传给订阅方。
import urllib
from flask_debugtoolbar import DebugToolbarExtension
from flask import Flask, render_template, Response, jsonify, session, escape, request, redirect, url_for
from serversess import RedisSessionInterface
from blinker import Namespace #导入模块
my_signals=Namespace() #创建命名空间，作用是可创建并存储多个信号发射对象
model_saved=my_signals.signal('model_saved') #创建一个信号发射对象，并赋予它一个<strong>独一无二</strong>的名字
app = Flask(__name__)
app.session_interface = RedisSessionInterface() #設置服務器端的session.默認情況下是client-side的session.
toolbar = DebugToolbarExtension(app)
app.config.from_object('settings')
class JSONResponce(Response):
@classmethod
def force_type(cls, response, environ=None):
if isinstance(response, dict):
response = jsonify(response)
return super(JSONResponce, cls).force_type(response, environ)
app.response_class = JSONResponce
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
session['username'] = request.form['username']
return redirect(url_for('index'))
return '''
<form method="post">
<p><input type=text name=username>
<p><input type=submit value=Login>
</form>'''
@app.route('/logout')
def logout():
# remove the username from the session if it's there
session.pop('username', None)
return redirect(url_for('index'))
@app.route('/')
def index():
if 'username' in session:
model_saved.send(app, data1='Login Signal', data2={'username': session.get('username')})
return 'Logged in as %s' % escape(session['username'])
return 'You are not logged in'
@app.route('/item/<int:id>/')
def item(id):
return {'Item number': 'item %d' % id}
@app.errorhandler(404)
def not_found(error):
return render_template('error.html'), 404
#接收信号
@model_saved.connect_via(app) #装饰器，接收app通过model_saved发送而来的信号
def signal_recv(app, data1, data2): #第一个参数app代表发送者，后面的参数为接收到的数据的键所对应的值
print('信号接收函数:{0}，{1}'.format(data1, data2))
pass
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=app.debug)
