flask信号

Posted meng0410

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了flask信号相关的知识,希望对你有一定的参考价值。

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()

 

以上是关于flask信号的主要内容,如果未能解决你的问题,请参考以下文章

python Flask - 数据库片段

Flask信号管理

Flask 信号机制 (signals)

Flask--信号

flask-信号

flask_day05:信号 Django信号 flask-script sqlalchemy 创建操作数据表