flask---信号

内容简介：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 学习笔记——信号

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。