Question

项目功能越复杂，代码量越大，就越需要做业务解耦。否则在其之上做开发和维护是很痛苦的，尤其是对于团队的新人。Flask 从 0.6 开始，通过 Blinker（http://bit.ly/28RwFDC ）提供了信号支持。信号就是在框架核心功能或者一些 Flask 扩展发生动作时所发送的通知，用于帮助你解耦应用。

Blinker 的使用

Blinker 目前还不是 Flask 的默认依赖，所以信号是不可用的。我们需要先安装它：

＞ pip install blinker

先了解下 Blinker（blinker_signal.py）：

from blinker import signal

started=signal('test-started')


def each(round):
    print 'Round{}!'.format(round)


def round_two(round):
    print 'Only{}'.format(round)


started.connect(each)
started.connect(round_two, sender=2)

for round in range(1, 4):
    started.send(round)

其中 connect 是订阅信号的方法。第二个参数是可选的，用于确定信号的发送端。“started.connect(round_two, sender=2)”表示值为 2 的时候才会接收。

send 是发送信号的方法。先看一下执行结果：

Round 1!
Round 2!
Only 2
Round 3!

可以设想，connect 和 send 这两个方法不放在一个文件中，它们通过 started 作为桥梁达到解耦的作用。

之前我们介绍过 Flask 的钩子，比如 before_request 和 after_request，这些钩子不需要 Blinker 库并且允许你改变请求对象（request）或者响应对象（response），而信号和钩子做的事情很像，只不过信号并不对请求对象和响应对象做改变，仅承担记录和通知的工作。

Flask 中内置的信号

Flask 可以发送 9 种信号，第三方的扩展中也可能会有额外的信号。而我们需要做的是添加对应的信号订阅。这里介绍常用的 6 种信号。

1．flask.template_rendered：模板渲染成功的时候发送，这个信号与模板实例 template、上下文的字典一起调用。

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)

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

3．flask.request_finished：在响应发送给客户端之前发送，可以传递 response。

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)

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

5．flask.request_tearing_down：在请求销毁时发送，它总是被调用，即使发生异常。

def close_db_connection(sender,**extra):
    session.close()

from flask import request_tearing_down
request_tearing_down.connect(close_db_connection, app)

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

自定义信号

在自己的应用中可以直接使用 Blinker 创建信号，现在定义一种对于上传大文件的信号：

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

当上传文件大于一个阈值的时候，就可以发送这个信号。

当编写一个 Flask 扩展并且想优雅地在未安装 Blinker 时退出，可以使用 flask.signals.Namespace - 在订阅信号的时候，如果发现未安装 Blinker 会抛出 RuntimeError。

信号订阅的高级用法

从 Blinker 1.1 开始可以用新的 connect_via() 装饰器订阅信号。比如使用上面介绍的 flask.appcontext_tearing_down：

@appcontext_tearing_down.connect_via(app)
def close_db_connection(sender,**extra):
    session.close()

也可以通过装饰器来用 connect 方法，比如下面的方法：

def each(round):
    print 'Round{}!'.format(round)


started.connect(each)

可以简写成：

@started.connect
def each(round):
    print 'Round{}!'.format(round)

Flask-Login 中的信号

Flask-Login 插件中带了 6 种信号（http://bit.ly/28PSm5F ），可以基于其中的信号做一些额外工作，比如基于 user_logged_in 来记录用户的登录次数和登录 IP 等。

我们先安装它：

＞ pip install flask-login

在 Flask-Login 中实现的发送信号代码是：

user_logged_in.send(current_app._get_current_object(), user=_get_user())

订阅这个信号可以了。现在我们实现一个功能齐备的登录例子（login_signal.py）。

初始化的部分和之前的例子很像，我们只展示不同的部分：

import flask_login

login_manager=flask_login.LoginManager()
login_manager.init_app(app)

password='123' #只要用户输入的密码是 123 就可以登录

flask_login 提供了 UserMixin，有一些用户相关的属性。

• is_authenticated：是否被验证。
• is_active：是否被激活。
• is_anonymous：是否是匿名用户。
• get_id()：获得用户的 id，并转换为 Unicode 类型。

可以在创建模型的时候继承 UserMixin：

class User(flask_login.UserMixin, db.Model):
    __tablename__='login_users'
    id=db.Column(db.Integer, primary_key=True)
    name=db.Column(db.String(128), nullable=False)
    login_count=db.Column(db.Integer, default=0)
    last_login_ip=db.Column(db.String(128), default='unknown')

db.create_all()

添加 user_logged_in 信号后，一登录就会触发 user_logged_in 信号，增加登录次数，并添加最近登录 IP：

    @flask_login.user_logged_in.connect_via(app)
def_track_logins(sender, user,**extra):
    user.login_count+=1
    user.last_login_ip=request.remote_addr
    db.session.add(user)
    db.session.commit()

使用 user_loader 装饰器的回调函数非常重要，它将决定 user 对象是否在登录状态：

@login_manager.user_loader
def user_loader(id):
    user=User.query.filter_by(id=id).first()
    return user

这个 id 参数的值是在 flask_login.login_user(user) 中传入的 user 的 id 属性。

现在添加登录的视图，如果是 GET 方法，只返回一个简单的表单：

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method=='GET':
        return '''
<form action='login' method='POST'>
    <input type='text' name='name' id='name' placeholder='name'></input>
    <input type='password' name='pw' id='pw' placeholder='password'></input>
    <input type='submit' name='submit'></input>
</form>
                  '''

    name=request.form.get('name')
    if request.form.get('pw')==password:
        user=User.query.filter_by(name=name).first()
        if not user:
            user=User(name=name)
            db.session.add(user)
            db.session.commit()
        flask_login.login_user(user)
        return redirect(url_for('protected'))

    return 'Bad login'

如果表单的密码是 123，就会跳转到视图函数 protected 上。显示登录的信息：

@app.route('/protected')
@flask_login.login_required
def protected():
    user=flask_login.current_user
    return 'Logged in as:{}|Login_count:{}|IP:{}'.format(
        user.name, user.login_count, user.last_login_ip)