Blinker提供了信号支持,信号就是在框架核心功能或者一些Flask扩展发生动作时所发送的通知,用于解耦应用。

1
2
3
4
5
6
from blinker import signal
started = signal('test-started')
def each(round):
print("round {}".format(round))
started.connect(each)
started.send(1)

connect和send方法不再一个文件中的时候,通过started作为桥梁达到解耦的作用。

Flask可以发送9种信号,第三方扩展中也会有额外的信号,需要做的是添加对应的信号订阅。

  • flask.template_rendered

模版渲染成功的时候发送,这个信号和模版实例template,上下文的字典一起调用。刚才有一个想法,就是可以用这些信号订阅来实现一个日志系统,将相应信号发生时候的参数都记录下来,这样方便调试和记录。

1
2
3
4
def log_template_renders(sender, template, context, **extra):
pass
from flask import template_rendered
template_rendered.connect(log_template_renders, app)
  • flask.request_started

建立请求上下文后,在请求处理开始之前发送,订阅者可以使用request之类的标准全局代理访问请求。

1
2
from flask import request_started
request_started.connect(log_request, app)
  • flask.request_finished
  • flask.got_request_exception 在请求处理中抛出异常时候发送,异常本身会通过exception传递到订阅函数
  • flask.request_tearing_down
  • flask.appcontext_tearing_down

自定义信号

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

当上传文件大于某个值,就可以使用send方法来发送这个信号。

信号订阅的高级用法

可以使用connect_via()装饰器订阅信号了

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

装饰器来使用connect方法,和上面的是一个作用

1
2
3
@started.connect
def each(round):
print("round {}".format(round))

flask-login中的信号

中有6种信号,可以基于其中信号做一些额外的工作,比如使用user_logged_in来记录用户的登录次数和登录IP

其中flask-login中发送信号的代码是:user_logged_in.send(current_app._get_current_object(), user=_get_user())

所以只要订阅这个信号,每次登录的时候就会触发订阅这个信号的函数,然后就执行这个函数:

1
2
3
4
5
6
@flask_login.user_logged_in.connect_via(app)
def _track_logins(sender, user, **extra):
user.login_count += 1
trueuser.last_login_ip = request.remote_addr
truedb.session.add(user)
truedb.session.commit()

实战应用

大概是这样的,每次在我发表新文章或者更新了一篇文章后,或者是更新了网站历史,发布了新版本的时候,都会连接新浪微博,然后发一条微博。这个机制中就用到了信号,实现了不同业务之间的解耦。很好用!

其中使用到了微博的API,或者你也可以直接通过模拟登录的方式。需要使用到一个第三方的包weibo,你可以点击这里找到它。

发送微博的代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from .weibo import Client
import random
APP_KEY = ''
APP_SECRET = ''
CALLBACK_URL = 'https://api.weibo.com/oauth2/default.html' # callback url
USERNAME = ''
PASSWORD = ''
def gen_location():
# -90 - +90
# -180 - +180
lat = random.randrange(-9000, 9000) / 100
long = random.randrange(-18000, 18000) / 100
return lat, long
def post_new(status):
c = Client(APP_KEY, APP_SECRET, CALLBACK_URL, username=USERNAME, password=PASSWORD)
location = gen_location()
# f = open(filename, 'rb')
c.post('statuses/update', status=status, lat=location[0], long=location[1])
# f.close()

接下来参考上面讲的信号理论内容来实践啦,首先创建signal,然后在发表新文章,更新文章或者历史的时候,来一句这个new_post.send(p)

signal定义和函数绑定如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from blinker import signal
from weiboRobot import post_new
new_post = signal('new_post')
update_post = signal('update_post')
new_history = signal('new_history')
@new_post.connect
def new_post_weibo(post):
status = "SkyRover 发表了新的文章《{}》, 快来围观吧! http://skyrover.me/post/{}".format(post.title, post.id)
post_new(status)
@update_post.connect
def update_post_weibo(post):
status = "SkyRover 更新文章《{}》, 快来围观吧! http://skyrover.me/post/{}".format(post.title, post.id)
post_new(status)
@new_history.connect
def new_history_weibo(history):
status = 'SkyRover 发布了新版本<{}>\n{}\n快来围观吧! http://skyrover.me'.format(history.title, history.body)
post_new(status)

参考自:Python Web 开发实战 – 董伟明