Flask信号管理
Posted Flask学习笔记
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Flask信号管理相关的知识，希望对你有一定的参考价值。
Flask信号机制
1.什么是Flask信号
Flask从0.6版本开始,信号功能有插件blinker来提供,这个插件必须安装才能开启信号功能,它的安装很简单,如下:$ pip install blinker那什么是信号呢?它指的是当核心框架或另一个
Flask扩展中发生动作时,信号通过发送通知来帮助解耦应用,也就是信号允许某个发送者通知接受者有事情发生.按照官网的说法,信号的使用目的是通知接受者,不鼓励接受者去修改数据,它的更多应用是在记录发生的这些事情,比如记录日志文件等等.在
Flask中所有与信号相关的函数定义在flask/signals.py文件中.参考
2.blinker
blinker是一个基于Python的信号库.支持一对一,一对多的订阅发布模式,支持发送任意大小数据,并且线程安全.根据
blinker可以自定义一个信号,它基本包括以下步骤:
创建信号 订阅信号 触发信号 首先介绍几个相关的类:
>>> import blinker
>>> help(blinker.Namespace)
A mapping of signal names to signals. 信号名称和信号的映射,命名空间,为了防止多人编写时,变量名称重复的问题
class Namespace(dict):
def signal(self, name, doc=None):
try:
return self[name]
except KeyError:
return self.setdefault(name, NamedSignal(name, doc))它的创建类似于:
from blinker import Namespace
# 1.生成一个Namespace命名空间,这是为了防止多人编写时定义了相同的变量名称
myspace = Namespace()
# signal中指定信号的名称
test_signle_1=myspace.signal('test_signle_1')接下来需要指定订阅信号,需要使用
connect()方法来指定订阅信号的函数,已经要执行的操作>>> help(blinker.Namespace().signal('test').connect)
connect(receiver, sender=ANY, weak=True)
receiver:指定一个可调用的函数,必须写,这个函数第一个参数必须是sender,用来指定订阅者,后面可以接任意的参数,这些参数由 send()方法发送
sender:指定订阅者,可以不写
Connect *receiver* to signal events sent by *sender*.它的使用类似与:
from blinker import Namespace
# 1.生成一个Namespace命名空间,这是为了防止多人编写时定义了相同的变量名称
myspace = Namespace()
# signal中指定信号的名称
test_signle_1=myspace.signal('test_signle_1')
# 2.订阅一个信号
def got_sender(sender):
print('sender %r' % sender)
test_signle_1.connect(got_sender)最后触发信号,可以传递一些参数给自定义的
receiver方法>>> help(blinker.Namespace().signal('test').send)
send(*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.它的使用是:
from blinker import Namespace
# 1.生成一个Namespace命名空间,这是为了防止多人编写时定义了相同的变量名称
myspace = Namespace()
# signal中指定信号的名称
test_signle_1=myspace.signal('test_signle_1')
# 2.订阅一个信号
def got_sender(sender):
print('sender %r' % sender)
test_signle_1.connect(got_sender)
# 3.触发信号
test_signle_1.send()以上是没有指定订阅者的,这里可以指定一个订阅者,并返回一些数据给自定义的函数
from blinker import Namespace
# 1.生成信号
my_space = Namespace()
test_signal_2=my_space.signal('test_signal_2')
# 2.订阅信号
class order():
def __init__(self,name):
self.name = name
def receiver(sender,name):
print('sender:{},args:{}'.format(sender,name))
test_signal_2.connect(receiver,order)
# 3.触发信号
test_signal_2.send(order,name='test')执行
❯ python single_test2.py
sender:<class '__main__.order'>,args:test
3.Flask 内置信号与blinker 关联
查看
flask/signals.pyfrom blinker import Namespace
_signals = Namespace()
template_rendered = _signals.signal("template-rendered")
before_render_template = _signals.signal("before-render-template")
request_started = _signals.signal("request-started")
request_finished = _signals.signal("request-finished")
request_tearing_down = _signals.signal("request-tearing-down")
got_request_exception = _signals.signal("got-request-exception")
appcontext_tearing_down = _signals.signal("appcontext-tearing-down")
appcontext_pushed = _signals.signal("appcontext-pushed")
appcontext_popped = _signals.signal("appcontext-popped")
message_flashed = _signals.signal("message-flashed")
Flask中内置了以上的信号,他们的对应是
信号 使用 template_rendered成功渲染模板后使用此信号. before_render_template渲染模板之前发送信号 request_started发送请求之前,发送信号 request_finished响应发送到客户端之前发送信号 got_request_exception请求处理期间发生异常发送此信号 request_tearing_down请求上下文消除时发送信号,即是有异常 appcontext_pushed推送应用程序上下文发送信号 appcontext_popedpop应用上下文发送信号message_flashed应用程序闪烁消息时发送此信号 可以看到
flask/signal.py中只是创建了信号,信号的订阅者,和触发者都定义在渲染模板相关函数中.# flask/temlating.py
def render_template(template_name_or_list, **context):
ctx = _app_ctx_stack.top
ctx.app.update_template_context(context)
return _render(
ctx.app.jinja_env.get_or_select_template(template_name_or_list),
context,
ctx.app,
)
def _render(template, context, app):
before_render_template.send(app, template=template, context=context)
rv = template.render(context)
template_rendered.send(app, template=template, context=context)
return rv
1.几个常用的信号
template_rendered: 渲染模板后的信号# app.py
from flask import Flask,render_template,template_rendered
app = Flask(__name__)
def template_rendered_func(sender,template,context):
print('sender:{}'.format(sender))
print('context:{}'.format(context))
print('template:{}'.format(template))
template_rendered.connect(template_rendered_func,app)
@app.route('/')
def index():
return render_template('index.html')
if __name__ == '__main__':
app.run()
"""
sender:<Flask 'app'>
context:{'g': <flask.g of 'app'>, 'request': <Request 'http://127.0.0.1:5000/' [GET]>, 'session': <NullSession {}>}
template:<Template 'index.html'>
"""从上面可以看到,手动实现了信号的订阅,因为信号的生成定义在了
flask/signals.py中,信号的触发定义在了flask/templating.py文件中,这里只需要手动的使用信号的订阅者和触发的函数就可以了.
got_request_exception:在请求处理视图函数时,发生异常可以调用这个信号,它定义在flask/app.py中.这个函数很有用,可以利用它来记录视图函数异常日志,# coding=utf-8
from flask import Flask,render_template,template_rendered,got_request_exception
app = Flask(__name__)
def template_rendered_func(sender,template,context):
print('sender:{}'.format(sender))
print('context:{}'.format(context))
print('template:{}'.format(template))
template_rendered.connect(template_rendered_func,app)
# 处理接受视图函数异常
def got_request_exception_func(sender,exception):
with open('exception.txt','a+') as fp:
fp.write('{sender}:{exception} \r\n'.format(sender=sender,exception=exception))
got_request_exception.connect(got_request_exception_func,app)
@app.route('/')
def index():
num = 1/0
return render_template('index.html')
if __name__ == '__main__':
app.run(debug=True)访问
127.0.0.1:5000后,会在当前文件夹下生成exception.txt❯ cat exception.txt
<Flask 'app'>:division by zero更多的方法参照官网
4.利用blinker 设置登录日志信号
- END -项目
.
├── app.py
├── app_siangels.py
├── log.txt
├── static
└── templates
├── index.html
└── login.html
app.py# coding=utf-8
from flask import Flask,render_template,request,g
from app_signals import login_signal
app = Flask(__name__)
app.config.update({
'DEBUG':True,
'TEMPLATE_AUTO_RELOAD':True
})
@app.route('/')
def index():
return render_template('index.html')
# 假设数据库文件
login = {'username':'admin','password':'123'}
@app.route('/login/',methods=['GET','POST'])
def login():
if request.method == 'GET':
return render_template('login.html')
else:
username = request.form.get('username')
g.username = username
password = request.form.get('password')
print(username,password)
if username == 'admin' and password == '123':
login_signal.send(app)
return 'login successful'
else:
return render_template('login.html')
if __name__ == '__main__':
app.run()
app_signals.pyfrom blinker import Namespace
from flask import request,g
from datetime import datetime
# 1.创建信号
myspace = Namespace()
login_signal = myspace.signal('login_signal')
# 2.响应信号
def login_log(sender):
username= g.username
now = datetime.now()
ip = request.remote_addr
log = "用户名:{},登录时间:{},ip地址:{}.\r\n".format(username,now,ip)
with open('log.txt','a+') as fp:
fp.write(log)
login_signal.connect(login_log)
html文档<!--login.html-->
<form action="" method="post">
<table>
<tr>
<td>用户名:</td>
<td><input type="text" name='username'></td>
</tr>
<tr>
<td>密码:</td>
<td><input type="password" name='password'></td>
</tr>
<tr>
<td><input type="submit" value="Login"></td>
</tr>
</table>
</form>访问即可得到日志文件
log.txt
以上是关于Flask信号管理的主要内容，如果未能解决你的问题，请参考以下文章