找到FLASK内部隐藏的所有钩子

• 0 钩子函数的作用
• 1 flask 运行时处理请求流程
• 2 app装饰器小结
• 3 信号小结
• 4 应用案例

当下的开发基本上都是基于框架做的，很少有重复造轮子的，所有的框架，不管是什么语音，都会在某些关键节点预留一些钩子函数，框架使用者通过这些钩子函数定制一些符合特定业务的功能。

抛开语言层面，任何web框架的大体逻辑大差不差，基本上都是封装Http请求，匹配路由，找到视图函数， 使用视图函数的返回值封装Http响应。在web框架中有几个关键节点，分别是每次接收请求前，匹配路由后，进入视图函数前，视图函数后等，FLASK内部就为用户设置了这些钩子函数，下面我就会从源码的角度看看他到底是怎么实现的，通过这个流程走下来，在使用flask做web开发的时候，如果需要扩展一些功能，我们就知道应该在哪扩展。

flask的内部通过两种方式实现的钩子函数，一种是Flask类提供的函数，另一种是通过信号

1. class Flask: _ _ call _ _方法

2. class Flask: wsgi_app 方法

   class Flask:
       def wsgi_app(self, environ, start_response):	
   		ctx = self.request_context(environ)
           error = None
           try:
               try:
                   ctx.push()
                   response = self.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)
   

   首先封装请求上下文对象，将请求数据封装到请求上下文对象中，请求上下文封装后执行push方法，将请求上下文对象入栈

   class RequestContext(object):
   	def push(self):
           app_ctx = _app_ctx_stack.top
           if app_ctx is None or app_ctx.app != self.app:
               app_ctx = self.app.app_context()
               app_ctx.push()
               self._implicit_app_ctx_stack.append(app_ctx)
           else:
               self._implicit_app_ctx_stack.append(None)
   
           if hasattr(sys, 'exc_clear'):
               sys.exc_clear()
   
           _request_ctx_stack.push(self)
   
           if self.session is None:
               session_interface = self.app.session_interface
               self.session = session_interface.open_session(
                   self.app, self.request
               )
   
               if self.session is None:
                   self.session = session_interface.make_null_session(self.app)
   

   从代码可以看出来，在推送请求上下文之间，flask会首先检查是否有应用上下文对象，如果没有应用上下文，先把应用上下文入栈Line:6：

   class AppContext(object):
   	def push(self):
           """Binds the app context to the current context."""
           self._refcnt += 1
           if hasattr(sys, 'exc_clear'):
               sys.exc_clear()
           _app_ctx_stack.push(self)
           appcontext_pushed.send(self.app)
   

   看上面代码块的第8行，我们找到了第一个flask信号，appcontext_pushed.send(self.app)，继续往后走：执行到wsgi函数的response = self.full_dispatch_request()这一句：

   class Flask:
   	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)
   

   ==self.try_trigger_before_first_request_functions()==这是flask的第二个扩展点：

   只在第一次请求前执行一次，一般用于一些初始化工作

   class Flask:
   	def try_trigger_before_first_request_functions(self):
           if self._got_first_request: # 默认是False 
               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 # 将__got_first_request 设置为True,下次在调用这个函数直接return
   	@setupmethod
       def before_first_request(self, f):
           # 使用这个函数装饰，就会把函数引用放到这个列表中
           self.before_first_request_funcs.append(f)
           return f
   

   继续往后走：==request_started.send(self)==第三个扩展点，开始处理请求。

   class Flask:
   	def preprocess_request(self):
           bp = _request_ctx_stack.top.request.blueprint
   
           funcs = self.url_value_preprocessors.get(None, ())
           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)
   
           funcs = self.before_request_funcs.get(None, ())
           if bp is not None and bp in self.before_request_funcs:
               funcs = chain(funcs, self.before_request_funcs[bp])
           for func in funcs:
               rv = func()
               if rv is not None:
                   return rv
   

   进入视图函数之前有两个扩展点：

   url_value_preprocessors里面的函数会被遍历执行，返回值无效，有两个参数分别是request.endpoint, request.view_args

   class Flask:
   	@setupmethod
       def url_value_preprocessor(self, f):
           # setdefault是字典一个特别好用的方法，给None设置一个默认值为空列表，返回值也是这个默认值，直接将被装饰的函数放到这个列表下
           self.url_value_preprocessors.setdefault(None, []).append(f)
           return f
   

   before_request_funcs，返回值会直接替代视图函数内的返回值

   class Flask: 
   	@setupmethod
       def before_request(self, f):
           self.before_request_funcs.setdefault(None, []).append(f)
           return 
   

   继续执行，如果before_request没有返回值执行： self.dispatch_request()

    class Flask:
    	def dispatch_request(self):
           # 我不理解为什么这里面不直接用request呢
           req = _request_ctx_stack.top.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)
   

   继续往下，使用视图函数的返回值，或者before_request函数的返回值作为http的返回值封装响应体

   class Flask:
   	def finalize_request(self, rv, from_error_handler=False):
           # make_response把rv里面的值做一些处理，满足http协议的规范
           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 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]))
           for handler in funcs:
               response = handler(response)
           if not self.session_interface.is_null_session(ctx.session):
               self.session_interface.save_session(self, ctx.session, response)
           return response
   
   

   _after_request_functions这个扩展点被添加到了请求上下文对象中，针对每次单次请求结束后做的事情：

   源码中的注释写的很清楚了，这里不多解释了。

   def after_this_request(f):
       """Executes a function after this request.  This is useful to modify
       response objects.  The function is passed the response object and has
       to return the same or a new one.
   
       Example::
   
           @app.route('/')
           def index():
               @after_this_request
               def add_header(response):
                   response.headers['X-Foo'] = 'Parachute'
                   return response
               return 'Hello World!'
   
       This is more useful if a function other than the view function wants to
       modify a response.  For instance think of a decorator that wants to add
       some headers without converting the return value into a response object.
   
       .. versionadded:: 0.9
       """
       _request_ctx_stack.top._after_request_functions.append(f)
       return f
   

   走到这，wsgi_app函数中的try代码块下面的代码算是执行完了

   跳过异常处理的代码块，走到finally块ctx.auto_pop(error)

   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]
                   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)
   	
       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)
   
       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)
       
       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)
       
       @setupmethod
       def teardown_request(self, f):
           self.teardown_request_funcs.setdefault(None, []).append(f)
           return f
   
       @setupmethod
       def teardown_appcontext(self, f):
           self.teardown_appcontext_funcs.append(f)
           return f
   

   class APPContext:
   	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]
                   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)
   

   do_teardown_request 在应用上下文被弹出时候执行

   do_teardown_appcontext 在应用上下文被弹出时候执行

   request_tearing_down 请求上下文弹出信号，传递了app对象和error

   appcontext_tearing_down 应用上下文弹出信号，传递了app对象和error

   appcontext_popped 应用上下文已经完全弹出信号，传递了app对象

from datetime import datetime

from flask import Flask, request, g

app = Flask(__name__)


@app.before_first_request
def before_first_request():
    """ 第一次请求前执行该函数，一般用于初始化信息 """
    print(datetime.now(), '打开数据库链接')


@app.url_value_preprocessor
def handle_url_value(view_endpoint, view_args):
    """ 函数在before_request前被触发 """
    print(datetime.now(), view_endpoint, view_args)


@app.before_request
def before_request():
    """ 每次请求前都会执行，一般用于权限管理，返回值会直接覆盖掉视图函数 """
    print(datetime.now(), '每次请求前都会被执行')
    # return 'xxxx' 慎用return


@app.route('/')
def index(nid:int):
    print(datetime.now(), '视图函数执行')
    from flask.ctx import after_this_request
    @after_this_request
    def hanlder_response(response):
        response.set_cookie('name', 'xx')
        response.headers['xxx'] = 111
        return response

    return f'Hello {nid}'


@app.teardown_request
def exe_at_req_ctx_pop(exc):
    """ 在请求上下文被弹出时执行 """
    print(datetime.now(), '请求上下文弹出')


@app.teardown_appcontext
def exe_at_app_ctx_pop(exc):
    """ 在应用上下文被弹出时执行 """
    print(datetime.now(), '应用上下文弹出')

    

"""
 * Running on http://0.0.0.0:8000/ (Press CTRL+C to quit)
2022-05-31 11:16:31.241793 打开数据库链接
2022-05-31 11:16:31.241793 index {'nid': 1}
2022-05-31 11:16:31.242793 每次请求前都会被执行
2022-05-31 11:16:31.242981 视图函数执行
2022-05-31 11:16:31.242981 请求上下文弹出
2022-05-31 11:16:31.242981 应用上下文弹出
127.0.0.1 - - [31/May/2022 11:16:31] "GET /1 HTTP/1.1" 200 -

"""

after_this_request设置的cookie和响应头

补充关于模板渲染的两个信号 templating.py

def _render(template, context, app):
    """Renders the template and fires the signal"""

    before_render_template.send(app, template=template, context=context)
    rv = template.render(context)
    template_rendered.send(app, template=template, context=context)
    return rv

before_render_template

template_rendered.send

@app.route('/')
def index(name):
    name = name
    #res = 1 / 0
    return render_template('index.html', name=name)


@signals.request_started.connect
def request_started(*args, **kwargs):
    print(args)
    print(kwargs)
    print(datetime.now(), '请求开始了')


@signals.request_finished.connect
def request_finished(*args, **kwargs):
    print(args)
    print(kwargs)
    print(datetime.now(), '请求结束了')


@signals.request_tearing_down.connect
def request_tearing_down(*args, **kwargs):
    print(args)
    print(kwargs)
    print(datetime.now(), '请求上下文弹出')


@signals.appcontext_tearing_down.connect
def appcontext_tearing_down(*args, **kwargs):
    print(args)
    print(kwargs)
    print(datetime.now(), '应用上下文弹出')


@signals.appcontext_pushed.connect
def _appcontext_pushed(*args, **kwargs):
    print(args)
    print(kwargs)
    print(datetime.now(), '应用上下文入栈')


@signals.appcontext_popped.connect
def _appcontext_popped(*args, **kwargs):
    print(args)
    print(kwargs)
    print(datetime.now(), '应用上下文出栈')


@signals.before_render_template.connect
def before_render_template(*args, **kwargs):
    print(args)
    print(kwargs)
    print(datetime.now(), '渲染模板前执行')


@signals.template_rendered.connect
def template_rendered(*args, **kwargs):
    print(args)
    print(kwargs)
    print(datetime.now(), '渲染模板完成后执行')


@signals.got_request_exception.connect
def got_request_exception(*args, **kwargs):
    print(args)
    print(kwargs)
    print(datetime.now(), '请求出错了')
    """
    (,)
    {'exception': ZeroDivisionError('division by zero')}
    2022-05-31 12:03:30.288377 请求出错了
    """
    
"""
(,)
{}
2022-05-31 12:04:33.326370 应用上下文入栈
(,)
{}
2022-05-31 12:04:33.327374 请求开始了
(,)
{'template':