内容简介：flask---信号
一、flask-信号
Flask框架中的信号基于blinker
flask-信号的安装
pip3 install blinker
10个内置信号
1. 内置信号
10个信号：
2. request_started = _signals.signal('request-started') # 请求到来前执行
5. request_finished = _signals.signal('request-finished') # 请求结束后执行
3. before_render_template = _signals.signal('before-render-template') # 模板渲染前执行
4. template_rendered = _signals.signal('template-rendered') # 模板渲染后执行
2/3/4/5或不执行 got_request_exception = _signals.signal('got-request-exception') # 请求执行出现异常时执行
6. request_tearing_down = _signals.signal('request-tearing-down') # 请求执行完毕后自动执行（无论成功与否）
7. appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 请求上下文执行完毕后自动执行（无论成功与否）
1. appcontext_pushed = _signals.signal('appcontext-pushed') # 请求app上下文push时执行
8. appcontext_popped = _signals.signal('appcontext-popped') # 请求上下文pop时执行
message_flashed = _signals.signal('message-flashed') # 调用flask在其中添加数据时，自动触发
二、信号源码这整个流程源码
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ) #实例化context这对象
ctx.push()
error = None
try:
try:
#执行：before装饰器函数，视图函数，after装饰器函数
response = self.full_dispatch_request()
except Exception as e: #（6.请求执行出现异常是执行got_request_exception）
error = e
response = self.handle_exception(e)
except:
error = sys.exc_info()[1]
raise
#将处理的内容返回给浏览器
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
#结束
ctx.auto_pop(error) #（ctx是requestContext的对象，执行ctx的pop方法）
def wsgi_app
def push(self):
app_ctx = _app_ctx_stack.top
if app_ctx is None or app_ctx.app != self.app:
#app_ctx=AppContext(self.app) (实例化做完开始执行 app_ctx.push()中的app_context中的push)
app_ctx = self.app.app_context() #把app放到队列放到Local中
#触发执行appcontext_push的方法（1.第一个信号已经触发被执行）
app_ctx.push()
self._implicit_app_ctx_stack.append(app_ctx) #（app_ctx是AppContext创建的对象）
else:
self._implicit_app_ctx_stack.append(None)
if hasattr(sys, 'exc_clear'):
sys.exc_clear()
#将RequestContext对象，添加到Local中
_request_ctx_stack.push(self)
# Open the session at the moment that the request context is
# available. This allows a custom open_session method to use the
# request context (e.g. code that access database information
# stored on `g` instead of the appcontext).
self.session = self.app.open_session(self.request)
if self.session is None:
self.session = self.app.make_null_session()
def push(
def full_dispatch_request(self):
#执行：@before_first_request装饰的所有函数
self.try_trigger_before_first_request_functions()
try:
request_started.send(self) #（2.视图函数没执行，请求刚进来执行 request_started.send(self)）
rv = self.preprocess_request()
if rv is None:
#执行视图函数
#执行session
rv = self.dispatch_request() #（执行视图函数）
#如果有模板先执行
# 3. before_render_template = _signals.signal('before-render-template') # 模板渲染前执行
# 4. template_rendered = _signals.signal('template-rendered') # 模板渲染后执行
#执行完这两个模板后在执行下面的
except Exception as e:
rv = self.handle_user_exception(e)
#执行：@after_request装饰所有的函数，保存session
return self.finalize_request(rv)
def full_dispatch_request
def process_response(self, response):
ctx = _request_ctx_stack.top
bp = ctx.request.blueprint
funcs = ctx._after_request_functions
if bp is not None and bp in self.after_request_funcs:
funcs = chain(funcs, reversed(self.after_request_funcs[bp]))
if None in self.after_request_funcs:
funcs = chain(funcs, reversed(self.after_request_funcs[None]))
#执行 @after_request装饰的所有的函数
for handler in funcs:
response = handler(response)
#最后处理session
if not self.session_interface.is_null_session(ctx.session):
self.save_session(ctx.session, response)
return response
def process_response
def finalize_request(self, rv, from_error_handler=False):
response = self.make_response(rv)
try:
response = self.process_response(response)
request_finished.send(self, response=response) #（5.视图函数执行完，请求结束后执行request_finished.send）
except Exception:
if not from_error_handler:
raise
self.logger.exception('Request finalizing failed with an '
'error while handling an error')
return response
def finalize_request
def auto_pop(self, exc):
if self.request.environ.get('flask._preserve_context') or \
(exc is not None and self.app.preserve_context_on_exception):
self.preserved = True
self._preserved_exc = exc
else:
self.pop(exc)
def auto_pop
def pop(self, exc=_sentinel):
app_ctx = self._implicit_app_ctx_stack.pop() （#app_ctx是AppContext(self.app)对象））
try:
clear_request = False
if not self._implicit_app_ctx_stack:
self.preserved = False
self._preserved_exc = None
if exc is _sentinel:
exc = sys.exc_info()[1]
self.app.do_teardown_request(exc)
# If this interpreter supports clearing the exception information
# we do that now. This will only go into effect on Python 2.x,
# on 3.x it disappears automatically at the end of the exception
# stack.
if hasattr(sys, 'exc_clear'):
sys.exc_clear()
request_close = getattr(self.request, 'close', None)
if request_close is not None:
request_close()
clear_request = True
finally:
rv = _request_ctx_stack.pop()
# get rid of circular dependencies at the end of the request
# so that we don't require the GC to be active.
if clear_request:
rv.request.environ['werkzeug.request'] = None
# Get rid of the app as well if necessary.
if app_ctx is not None:
app_ctx.pop(exc)
assert rv is self, 'Popped wrong request context. ' \
'(%r instead of %r)' % (rv, self)
def pop
def do_teardown_request(self, exc=_sentinel):
if exc is _sentinel:
exc = sys.exc_info()[1]
funcs = reversed(self.teardown_request_funcs.get(None, ()))
bp = _request_ctx_stack.top.request.blueprint
if bp is not None and bp in self.teardown_request_funcs:
funcs = chain(funcs, reversed(self.teardown_request_funcs[bp]))
for func in funcs:
func(exc)
request_tearing_down.send(self, exc=exc) #( 6.request_tearing_down请求执行完毕后自动执行（无论成功与否）)
def do_teardown_request
def do_teardown_appcontext(self, exc=_sentinel):
if exc is _sentinel:
exc = sys.exc_info()[1]
for func in reversed(self.teardown_appcontext_funcs):
func(exc)
appcontext_tearing_down.send(self, exc=exc) (#7.appcontext_tearing_down请求上下文执行完毕后自动执行（无论成功与否）)
def do_teardown_appcontext
def pop(self, exc=_sentinel):
try:
self._refcnt -= 1
if self._refcnt <= 0:
if exc is _sentinel:
exc = sys.exc_info()[1]
self.app.do_teardown_appcontext(exc)
finally:
rv = _app_ctx_stack.pop()
assert rv is self, 'Popped wrong app context. (%r instead of %r)' \
% (rv, self)
appcontext_popped.send(self.app) (#8. appcontext_popped # 请求上下文pop时执行)
def pop
三、自定义信号
from flask import Flask,flash
from flask.signals import _signals
app = Flask(__name__)
wh = _signals.signal('wh')
# 定义函数
def luominwen(*args,**kwargs):
print('罗姑娘',args,kwargs)
# 定义函数
def shaowei(*args,**kwargs):
print('少姑娘',args,kwargs)
# 将函数注册到request_started信号中： 添加到这个列表
wh.connect(luominwen)
wh.connect(shaowei)
@app.route('/index')
def index():
# 触发这个信号：执行注册到列表中的所有函数
# 发送短信，邮件，微信
wh.send(sender='xxx',a1=123,a2=456)
return "xx"
if __name__ == '__main__':
app.__call__
app.run()
问题：
特殊的装饰器和信号有什么区别？
- 装饰器返回值有意义
- 信号用于做什么呢？
- 降低代码之间的耦合
四、django内置的信号
Request/response signals
request_started # 请求到来前，自动触发
request_finished # 请求结束后，自动触发
got_request_exception # 请求异常后，自动触发
Model signals
pre_init # django的modal执行其构造方法前，自动触发
post_init # django的modal执行其构造方法后，自动触发
pre_save # django的modal对象保存前，自动触发
post_save # django的modal对象保存后，自动触发
pre_delete # django的modal对象删除前，自动触发
post_delete # django的modal对象删除后，自动触发
m2m_changed # django的modal中使用m2m字段操作第三张表（add,remove,clear）前后，自动触发
class_prepared # 程序启动时，检测已注册的app中modal类，对于每一个类，自动触发
Management signals
pre_migrate # 执行migrate命令前，自动触发
post_migrate # 执行migrate命令后，自动触发
Test signals
setting_changed # 使用test测试修改配置文件时，自动触发
template_rendered # 使用test测试渲染模板时，自动触发
Database Wrappers
connection_created # 创建数据库连接时，自动触发
需求：新浪面试，数据库12张表，每张表创建一条数据时，记录一条日志。
重写save方法或者用信号，找到信号注册一个函数
注意注册一个函数是：把这个注册的函数放在文件刚启动起来的位置，如：__init.py 里面
注册的函 数为：
def func(*args,**):
print(args,kwargs)
post_save.connect(func)
v1 = [11,22,33,44]
v2 = [1,4,7,5]
from itertools import chain
ff = []
for i in chain(v1,v2): #chain会把两个列表连接在一块
ff.append(i)
print(ff) #[11, 22, 33, 44, 1, 4, 7, 5]
chain的应用
flask信号本生自己没有，用的是别人的，并且这些信号通过装饰器全部可以代替了的，但是Django里面有些特殊的
就是那些model操作根本没有装饰器，就是同过内置的信号来完成的
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持 码农网
猜你喜欢:
- 树莓派基础-模拟信号和数字信号的区别
- xenomai内核解析之信号signal(二)---xenomai信号处理机制
- 没信号也可救你命：苹果新专利让iPhone无信号也可充当紧急信标
- 信号量
- Python信号处理
- APUE 学习笔记——信号
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
