05 Flask信号及源码分析

187次阅读
没有评论

共计 5229 个字符,预计需要花费 14 分钟才能阅读完成。

一. 信号简介

05 Flask 信号及源码分析

Flask 信号中文手册 : https://docs.pythontab.com/flask/flask0.10/signals.html

简而言之 : Flask 框架中的信号基于 blinker,其主要就是让开发者可是在 flask 请求过程中定制一些用户行为

二. 安装

pip3 install blinker

三. 内置信号(10 个)

1.request_started = _signals.signal('request-started')                # 请求到来前执行
2.request_finished = _signals.signal('request-finished')              # 请求结束后执行

3.before_render_template = _signals.signal('before-render-template')  # 模板渲染前执行
4.template_rendered = _signals.signal('template-rendered')            # 模板渲染后执行

5.got_request_exception = _signals.signal('got-request-exception')    # (1/2/3/ 4 请求执行出现异常时执行)

6.request_tearing_down = _signals.signal('request-tearing-down')      # 请求执行完毕后自动执行(无论成功与否)7.appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 应用上下文执行完毕后自动执行(无论成功与否)8.appcontext_pushed = _signals.signal('appcontext-pushed')            # 应用上下文 push 时执行
9.appcontext_popped = _signals.signal('appcontext-popped')            # 应用上下文 pop 时执行
10.message_flashed = _signals.signal('message-flashed')                # 调用 flask 在其中添加数据时,自动触发

四. 信号的使用步骤

from flask import Flask, signals

app = Flask(__name__)

# 1. 定义一个函数
def func1(*args, **kwargs):
    print('请求来了, 信号被触发', '\n', args, '\n', kwargs)


def func2(*args, **kwargs):
    print('请求走了, 信号被触发', '\n', args, '\n', kwargs)


# 2. 往信号中 (内置信号中的一个) 注册函数
signals.request_started.connect(func1)
signals.request_finished.connect(func2)


# 3. 触发信号(由于是内置信号,会自动触发,只要触发信号就会执行这个函数)
@app.route('/')
def index():
    return 'ok'

五. 自定义信号

from flask import Flask, signals
from blinker import Namespace
from flask.signals import _signals

app = Flask(__name__)

## 方式一
my_signals = Namespace()
# 定义了一个 custom 信号
custom = my_signals.signal('custom')

## 方式二
# 定义了一个 xxx 信号
xxx = _signals.signal('xxx')


def func3(*args, **kwargs):
    print('信号被触发', '\n', args, '\n', kwargs)


@app.route('/')
def index():
    # 往自定义信号中注册函数,send 触发信号
    custom.connect(func3)
    custom.send(name="shawn")

    # 往自定义信号中注册函数,send 触发信号
    xxx.connect(func3)
    xxx.send(name="song")
    return 'ok'


if __name__ == '__main__':
    app.run()
"""
信号被触发 
 (None,) 
 {'name': 'shawn'}

信号被触发 
 (None,) 
 {'name': 'song'}
"""

六. 源码分析

1.request_started = _signals.signal('request-started')

  • wsgi_app(self) 方法中的 def full_dispatch_request(self): 方法 request_started.send(self) 触发信号
def full_dispatch_request(self):
    self.try_trigger_before_first_request_functions()
    try:
        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)
    return self.finalize_request(rv)

由上面源码可以发现 request_started 信号在 before_first_request() 请求扩展之后触发, 在 befor_request() 之前触发

  • 示例
from flask import Flask, signals

app = Flask(__name__)

def func1(*args, **kwargs):
    print('信号被触发')

signals.request_started.connect(func1)

@app.route('/')
def index():
    return 'ok'

@app.before_first_request
def before_first():
    print('第一次请求触发')
@app.before_request
def before():
    print('请求触发')

if __name__ == '__main__':
    app.run()
"""
第一次请求触发
信号被触发
请求触发
"""

2.request_finished = _signals.signal('request-finished')

  • request_started 中的 return : 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)
        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

该信号在请求结束后执行

3.before_render_template = _signals.signal('before-render-template')

  • 渲染模板之前执行
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)
def _render(template, context, app):
    # 渲染之前触发
    before_render_template.send(app, template=template, context=context)
    rv = template.render(context)
    # 渲染之后触发
    template_rendered.send(app, template=template, context=context)
    return rv

4.template_rendered = _signals.signal('template-rendered')

  • 渲染模板之后执行
def _render(template, context, app):
    # 渲染之前触发
    before_render_template.send(app, template=template, context=context)
    rv = template.render(context)
    # 渲染之后触发
    template_rendered.send(app, template=template, context=context)
    return rv

5.got_request_exception = _signals.signal('got-request-exception')

  • 前四个请求出现异常时触发执行
  • wsgi_app() > response = self.full_dispatch_request() 只要这一句出现异常 > 触发 response = self.handle_exception(e) 的执行
def wsgi_app(self, environ, start_response):
     ...
     try:
         try:
             response = self.full_dispatch_request()
         except Exception as e:
             error = e
             response = self.handle_exception(e)
         return response(environ, start_response)
     finally:
         ...
  • self.handle_exception(e)
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())

6.request_tearing_down = _signals.signal('request-tearing-down')

  • 请求执行完毕后自动触发, 无论成功与否
  • wsgi_app() > self
  • .request_context(environ) > return RequestContext(self, environ) > pop() > self.app.do_teardown_request(exc)
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)  # 触发信号

7.appcontext_tearing_down = _signals.signal('appcontext-tearing-down')

  • 应用上下文执行完毕后自动执行, 无论成功与否
  • do_teardown_appcontext
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)  # 触发信号

...

.....

正文完
 
shawn
版权声明:本站原创文章,由 shawn 2023-06-16发表,共计5229字。
转载说明:除特殊说明外本站文章皆由CC-4.0协议发布,转载请注明出处。
评论(没有评论)