信号 | 独角兕大王

# 信号

Flask框架中的信号基于blinker，其主要就是让开发者可是在flask请求过程中定制一些用户行为。比如：在存储数据库之前，对每个请求的数据进行写入一条数据

# 准备工作

安装bliker第三方模块
```
pip3 install blinker
```

原因

from flask import Flask,signals  #导入signals
  #点击signals源码中，需要导入blinker模块
  #signals.py中
try:
    from blinker import Namespace
    signals_available = True
except ImportError:
    signals_available = False
    ...

template_rendered = _signals.signal("template-rendered")
before_render_template = _signals.signal("before-render-template")
request_started = _signals.signal("request-started")
request_finished = _signals.signal("request-finished")
request_tearing_down = _signals.signal("request-tearing-down")
got_request_exception = _signals.signal("got-request-exception")
appcontext_tearing_down = _signals.signal("appcontext-tearing-down")
appcontext_pushed = _signals.signal("appcontext-pushed")
appcontext_popped = _signals.signal("appcontext-popped")
message_flashed = _signals.signal("message-flashed")

# 内置信号

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在其中添加数据时，自动触发

# 信号使用简单演示

from flask import Flask,signals

app = Flask('app01')
app.debug = True

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

signals.request_started.connect(func)  # 使用connect将函数注册到信号中，也就是说请求进来就执行，send()是执行信号中的函数

@app.route('/',methods=['GET'])
def index():
    print('视图')
    return 'index'

if __name__ == '__main__':
    app.run()
 
 # 运行结果
 触发信号 (<Flask 'app01'>,) {}
 视图

小提示

1、.connect(func)  #是注册函数进信号
2、.send()  #是执行信号中的函数

# 信号在哪触发？怎么触发？源码分析

依旧查看app.__call__中的wsgi_app函数

 def wsgi_app(self, environ, start_response):
        ctx = self.request_context(environ) #将请求相关数据封装到RequestContext的对象ctx中，包括ctx.app、ctx.request、ctx.session
        error = None
        try:
            try:
                ctx.push() #将ctx、app_ctx通过LocalStack添加到local中，上下文管理
               
                response = self.full_dispatch_request() #寻找试图函数执行，将返回值给用户
            except Exception as e:
                error = e
                response = self.handle_exception(e) #将出错的响应给用户
            except:  # noqa: B001
                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、ctx.push进行了详细的分析，接着我们将继续往下进行分析，对full_dispatch_request()进行分析，点击查看源码

    def full_dispatch_request(self):
        self.try_trigger_before_first_request_functions() #用于处理before_first_request函数
        try:
            # 执行before_first_request
            request_started.send(self) #触发request_started信号里面的注册的函数（connect）
            
            #预处理请求，调用before_request函数，包括蓝图的before_request函数
            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)  #最后返回返回值

点击 try_trigger_before_first_request_functions() 函数

    def try_trigger_before_first_request_functions(self):
        if self._got_first_request:  #默认为Flase，第二次进来就为True了，因为已经改了
            return
        with self._before_request_lock: #添加了一个锁，是列表安全
            if self._got_first_request:
                return 
            for func in self.before_first_request_funcs: #循环添加的列表，并执行
                func()
            self._got_first_request = True #关门，第二次进行了就直接return

点击查看 self.preprocess_request() 函数的源码

    def preprocess_request(self):
        bp = _request_ctx_stack.top.request.blueprint

        funcs = self.url_value_preprocessors.get(None, ()) #处理url相关的
        if bp is not None and bp in self.url_value_preprocessors:
            funcs = chain(funcs, self.url_value_preprocessors[bp])
        for func in funcs:
            func(request.endpoint, request.view_args)
		
        # self.before_request_funcs = {} 是全局的一个字典
        funcs = self.before_request_funcs.get(None, ()) #向全局before_request_func字典获取函数
        if bp is not None and bp in self.before_request_funcs:# 注意：默认app存入before_request_funcs字典的的key是None，而蓝图存入的带有子的蓝图名字
            funcs = chain(funcs, self.before_request_funcs[bp]) #如果app和蓝图都有before_request函数，就是用chain链式函数，将函数进行排序，输出
        for func in funcs: #循环每个before_request函数
            rv = func()  #如果函数有返回值，就返回返回值
            if rv is not None:
                return rv

dispatch_request() 函数源码，该函数是之执行试图函数

    def dispatch_request(self):
        req = _request_ctx_stack.top.request #获得ctx中的request属性
        if req.routing_exception is not None:
            self.raise_routing_exception(req)
        rule = req.url_rule
        # if we provide automatic options for this URL and the
        # request came with the OPTIONS method, reply automatically
        if (
            getattr(rule, "provide_automatic_options", False)
            and req.method == "OPTIONS"
        ):
            return self.make_default_options_response()
        # otherwise dispatch to the handler for that endpoint，通过别名找到视图函数并执行
        return self.view_functions[rule.endpoint](**req.view_args)

如果我们使用了render_template() 函数，查看源码

def render_template(template_name_or_list, **context): #模板的参数，模板渲染
    ctx = _app_ctx_stack.top  # 获取到app_ctx中的的AppContext对象
    ctx.app.update_template_context(context)  #将传入的参数给模板中的响应的自定义函数
    return _render(  #调用_render函数
        ctx.app.jinja_env.get_or_select_template(template_name_or_list),
        context,
        ctx.app,
    )

2、点击update_template_context(),为模板中的使用的一些自定义函数，传递参数

2、查看_render函数源码：
def _render(template, context, app):
    #如果我们写的时候，谢了before_render_template.connect() 函数，就会执行下面一句
    before_render_template.send(app, template=template, context=context) 
    rv = template.render(context)  #将参数传入，进行模板渲染
    template_rendered.send(app, template=template, context=context) #模板渲染后执行，需要我们自己定义的template_rendered.connect()函数
    return rv

response(environ, start_response) 函数源码分析，该函数实则是执行了self.finalize_request(rv) （表示正常处理结果，不会被异常接获）、handle_exception() 函数捕获一异常返回的finalize_request(server_error, from_error_handler=True)函数，打开finalize_request()函数源码

    def finalize_request(self, rv, from_error_handler=False):
        response = self.make_response(rv)
        try:     
            response = self.process_response(response) #处理after_request函数，session_save函数
            request_finished.send(self, response=response) #执行请求结束执行的函数，需要我们自己使用connect注册
        except Exception:
            if not from_error_handler:
                raise
            self.logger.exception(
                "Request finalizing failed with an error while handling an error"
            )
        return response

# before_request 和 before_first_request 请求扩展源码分析

简单实例

from flask import Flask,signals

app = Flask('app01')
app.debug = True

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

signals.request_started.connect(func)  # 使用connect将函数注册到信号中，也就是说请求进来就执行

@app.before_first_request
def before_request():
    pass

@app.before_first_request  # 向全局before_first_request_funcs列表添加两个函数
def before_request2():
    pass

@app.before_request  # 向全局before_first_request_funcs列表添加两个函数
def before_request3():
    pass

@app.route('/',methods=['GET'])
def index():
    print('视图')
    return 'index'


if __name__ == '__main__':
    app.__call__
    app.run()

# 源码分析

点击before_reqeust装饰器函数，查看源码

 @setupmethod
 def before_first_request(self, f):
     self.before_first_request_funcs.append(f)  #将装饰的函数添加到全局的一个列表中
     return f #最后将函数返回，没有inner函数

点击before_request装饰器函数，查看源码

 @setupmethod
 def before_request(self, f):
     self.before_request_funcs.setdefault(None, []).append(f) #向全局的self.before_request_funcs字典中添加，添加空列表
     #得到的值字典：{None: ['1']}
     return f

# 梳理Flask信号流程

# 总结以上信号流程

1、请求进来执行：before_first_request
2、触发 reqeust_started 信号
3、before_request函数
4、如果视图函数中有模板渲染：
	渲染前的信号：before_render_template.send
				rv = template.render() #模板渲染
	渲染后的信号：template_rendered.send()
5、after_requset 函数
6、session.save_session()
7、触发request.finished信号

8、如果触发异常
	9、触发错误处理的信号：执行handle_exception()函数中的got_request_exception.send()函数
10、在执行auto_pop()函数中的do_teardown_request()函数中： 执行request_teardown.send()

# 自定义Flask信号

 #!/usr/bin/env python
 # -*- coding:utf-8 -*-
from flask import Flask, current_app, flash, render_template
from flask.signals import _signals
 
app = Flask(import_name=__name__)
 
 
# 自定义信号
xxxxx = _signals.signal('xxxxx')
 
def func(sender, *args, **kwargs):
    print(sender)
 
# 自定义信号中注册函数
xxxxx.connect(func)
 
 
@app.route("/x")
def index():
    # 触发信号
    xxxxx.send('123123', k1='v1')
    return 'Index'
 
 
if __name__ == '__main__':
    app.run()

← WTForms 应用上下文 →