05 Flask信号及源码分析

228次阅读

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

05 Flask 信号及源码分析

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

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

pip3 install blinker

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'}
"""

在 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()
"""
第一次请求触发
信号被触发
请求触发
"""

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

该信号在请求结束后执行

渲染模板之前执行

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

渲染模板之后执行

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

前四个请求出现异常时触发执行
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())

请求执行完毕后自动触发, 无论成功与否
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)  # 触发信号

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

...

.....

正文完