简单了解信号
Flask框架中的信号基于blinker，其主要就是让开发者可是在flask请求过程中定制一些用户行为。简单来说就是flask在列表里面，预留了几个空列表，在里面存东西。信号通过发送通知来帮助你解耦应用。简言之，信号允许某个发送者通知接收者有事情发生了；
1 | pip3 install blinker |
|---|
一、 内置信号
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | 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在其中添加数据时，自动触发 |
|---|
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在其中添加数据时，自动触发源码示例
代码语言：javascript复制class Flask(_PackageBoundObject):
def full_dispatch_request(self):
self.try_trigger_before_first_request_functions()
try:
# ############### 触发request_started 信号 ###############
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
response = self.make_response(rv)
response = self.process_response(response)
# ############### request_finished 信号 ###############
request_finished.send(self, response=response)
return response
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.make_response(self.handle_exception(e))
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)同上def render_template(template_name_or_list, **context):
"""Renders a template from the template folder with the given
context.
:param template_name_or_list: the name of the template to be
rendered, or an iterable with template names
the first one existing will be rendered
:param context: the variables that should be available in the
context of the template.
"""
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):
"""Renders the template and fires the signal"""
# ############### before_render_template 信号 ###############
before_render_template.send(app, template=template, context=context)
rv = template.render(context)
# ############### template_rendered 信号 ###############
template_rendered.send(app, template=template, context=context)
return rv同上class Flask(_PackageBoundObject):
def handle_exception(self, e):
exc_type, exc_value, tb = sys.exc_info()
# ############### got_request_exception 信号 ###############
got_request_exception.send(self, exception=e)
handler = self._find_error_handler(InternalServerError())
if self.propagate_exceptions:
# if we want to repropagate the exception, we can attempt to
# raise it with the whole traceback in case we can do that
# (the function was actually called from the except part)
# otherwise, we just raise the error again
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 handler(e)
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.make_response(self.handle_exception(e))
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)class AppContext(object):
def push(self):
"""Binds the app context to the current context."""
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]
# ############## 触发 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 push(self):
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop(top._preserved_exc)
app_ctx = _app_ctx_stack.top
if app_ctx is None or app_ctx.app != self.app:
# ####################################################
app_ctx = self.app.app_context()
app_ctx.push()
self._implicit_app_ctx_stack.append(app_ctx)
else:
self._implicit_app_ctx_stack.append(None)
if hasattr(sys, 'exc_clear'):
sys.exc_clear()
_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()
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.make_response(self.handle_exception(e))
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
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]
# ################## 触发 request_tearing_down 信号 ##################
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 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 flash(message, category='message'):
"""Flashes a message to the next request. In order to remove the
flashed message from the session and to display it to the user,
the template has to call :func:`get_flashed_messages`.
.. versionchanged:: 0.3
`category` parameter added.
:param message: the message to be flashed.
:param category: the category for the message. The following values
are recommended: ``'message'`` for any kind of message,
``'error'`` for errors, ``'info'`` for information
messages and ``'warning'`` for warnings. However any
kind of string can be used as category.
"""
# Original implementation:
#
# session.setdefault('_flashes', []).append((category, message))
#
# This assumed that changes made to mutable structures in the session are
# are always in sync with the session object, which is not true for session
# implementations that use external storage for keeping their keys/values.
flashes = session.get('_flashes', [])
flashes.append((category, message))
session['_flashes'] = flashes
# ############### 触发 message_flashed 信号 ###############
message_flashed.send(current_app._get_current_object(),
message=message, category=category)二、 自定义信号
第一步：创建信号
第二步：将函数注册到信号中： 添加到这个列表
第三步： 发送信号
第四步：运行
具体实现：可参考flask源码，写一个自定义信号
代码语言：javascript复制from flask import Flask,flash
from flask.signals import _signals
app = Flask(__name__)
xinhao = _signals.signal("xinhao")#创建信号
#定义函数
def wahaha(*args,**kwargs):
print("111",args,kwargs)
def sww(*args,**kwargs):
print("222",args,kwargs)
# 将函数注册到信号中,添加到这个列表
xinhao.connect(wahaha)
xinhao.connect(sww)
@app.route("/zzz")
def zzz():
xinhao.send(sender='xxx',a1=123,a2=456) #触发这个信号，执行注册到这个信号列表中的所有函数,此处的参数个数需要与定义的函数中的参数一致
return "发送信号成功"
if __name__ == '__main__':
app.run(debug=True)
#打印结果
# 111 (None,) {'sender': 'xxx', 'a1': 123, 'a2': 456}
# 222 (None,) {'sender': 'xxx', 'a1': 123, 'a2': 456}三、了解Django中的信号
点击查看
四、其他
1.chain的作用
代码语言：javascript复制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]2.特殊的装饰器（@app.before_first_request ；@app.before_request ； @app.after_request）和信号有什么区别？
- 触发信号是没有返回值的，写不写返回值都无所谓
- 特殊的装饰器对返回值是有意义的，当before_request有返回值时就不会执行后续视图函数了，没有返回值的时候才会执行后续函数，而after_request必须有返回值
所以特殊装饰器的功能比信号的功能强大
3.信用用于做什么？
-自定义一些没有返回值的操作 -降低代码之间的耦合
4.通过信号可以做权限吗？
- 本身是做不了的，要想做得用其他的机制配合着来使用，这样做的话会闲的很麻烦，所以我们选择中间件来做
