Flask框架中信号基于blinker,可以让开发人员在flask请求过程中定制一些用户行为
- 安装blinker:pip3 install blinker
- blinker提供的信号
request_started = _signals.signal(‘request-started‘) # 请求到来前执行
request_finished = _signals.signal(‘request-finished‘) # 请求结束后执行
before_render_template = _signals.signal(‘before-render-template‘) # 模板渲染前执行
template_rendered = _signals.signal(‘template-rendered‘) # 模板渲染后执行
got_request_exception = _signals.signal(‘got-request-exception‘) # 请求执行出现异常时执行
request_tearing_down = _signals.signal(‘request-tearing-down‘) # 请求执行完毕后自动执行（无论成功与否）
appcontext_tearing_down = _signals.signal(‘appcontext-tearing-down‘)# 请求上下文执行完毕后自动执行（无论成功与否）
appcontext_pushed = _signals.signal(‘appcontext-pushed‘) # 请求上下文push时执行
appcontext_popped = _signals.signal(‘appcontext-popped‘) # 请求上下文pop时执行
message_flashed = _signals.signal(‘message-flashed‘) # 调用flask在其中添加数据时，自动触发
class Flask(_PackageBoundObject):
def wsgi_app(self, environ, start_response)
ctx = self.request_context(environ)
ctx.push()
error = None
try:
try:
response = self.full_dispatch_request() #调用full_dispatch_request方法
except Exception as e:
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)
def full_dispatch_request(self):
self.try_trigger_before_first_request_functions() #执行@before_first_request这个装饰器装饰的函数
try:
##################触发request_started信号############################
request_started.send(self)
rv = self.preprocess_request() #执行@before_request这个装饰器装饰的函数
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
def finalize_request(self, rv, from_error_handler=False):
response = self.make_response(rv)
try:
response = self.process_response(response) #执行@after_request这个装饰器装饰的函数同时会保存session
###################触发request_finished信号###############
request_finished.send(self, response=response)
except Exception:
if not from_error_handler:
raise
self.logger.exception(‘Request finalizing failed with an ‘
‘error while handling an error‘)
return response
def _render(template, context, app):
####################触发before_render_template信号###############
before_render_template.send(app, template=template, context=context)
rv = template.render(context)
###################触发templated_rendered信号#####################
template_rendered.send(app, template=template, context=context)
return rv
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)
class Flask(_PackageBoundObject):
def wsgi_app(self, environ, start_response)
ctx = self.request_context(environ)
ctx.push()
error = None
try:
try:
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e) #调用handle_exception方法
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)
def handle_exception(self, e):
exc_type, exc_value, tb = sys.exc_info()
#################触发错误处理信号#############
got_request_exception.send(self, exception=e)
handler = self._find_error_handler(InternalServerError())
if self.propagate_exceptions:
if exc_value is e:
reraise(exc_type, exc_value, tb)
else:
raise e
self.log_exception((exc_type, exc_value, tb))
if handler is None:
return InternalServerError()
return self.finalize_request(handler(e), from_error_handler=True)
class AppContext(object):
def push(self):
self._refcnt += 1
if hasattr(sys, ‘exc_clear‘):
sys.exc_clear()
_app_ctx_stack.push(self)
######触发appcontext_pushed信号#########
appcontext_pushed.send(self.app)
def pop(self, exc=_sentinel):
"""Pops the app context."""
try:
self._refcnt -= 1
if self._refcnt <= 0:
if exc is _sentinel:
exc = sys.exc_info()[1]
#调用Flask类中的do_teardown_appcontext方法，触发appcontext_tearing_down信号
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信号#######
appcontext_popped.send(self.app)
class RequestContext(object):
def pop(self, exc=_sentinel):
app_ctx = self._implicit_app_ctx_stack.pop()
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]
#调用Flask类中调用do_teardown_request方法，触发 request_tearing_down信号
self.app.do_teardown_request(exc)
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()
if clear_request:
rv.request.environ[‘werkzeug.request‘] = None
if app_ctx is not None:
app_ctx.pop(exc)
assert rv is self, ‘Popped wrong request context. ‘ ‘(%r instead of %r)‘ % (rv, self)
class Flask(_PackageBoundObject):
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信号########
request_tearing_down.send(self, exc=exc)
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信号
appcontext_tearing_down.send(self, exc=exc)
自定义信号
一个信号需满足三个条件：有信号、信号中必须注册函数、请求触发函数
from flask import Flask,current_app,flash,render_template
from flask.signals import _signals
app=Flask(__name__)
# 自定义信号
sig=_signals.signal(‘sig‘)
def func(sender,*args,**kwargs):
print(sender)
# 给自定义信号中注册函数
sig.connect(func)
@app.route(‘/‘)
def hello_world():
# 触发信号
sig.send(‘hello world‘)
return ‘hello world‘
if __name__ == ‘__main__‘:
app.run()
