一、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 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 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 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 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 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 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 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_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 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时执行)
三、自定义信号
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]
flask信号本生自己没有，用的是别人的，并且这些信号通过装饰器全部可以代替了的，但是Django里面有些特殊的
就是那些model操作根本没有装饰器，就是同过内置的信号来完成的