flask---信号

2018-06-18 00:26:56来源:未知 阅读 ()

新老客户大回馈,云服务器低至5折

一、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操作根本没有装饰器,就是同过内置的信号来完成的

 

 

标签:

版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有

上一篇:Python:名片管理系统2.0(增加登录功能)

下一篇:Flask学习笔记--数据库(Python3.6)