Flask 源码阅读-正菜

278次阅读  |  发布于3年以前

flask项目大名鼎鼎，不需要多做介绍。我把它称之为python服务开发的TOP2项目，另外一个就是django。这两个项目各有千秋，各自有不同的应用场景，都需要深入理解，熟练掌握。本次源码选择的版本是 1.1.2，我会采用慢读法，尽自己最大努力把它讲透。本篇是第2篇-正菜，主要包括:

flask-app创建

还是先从flask-app的示例开始:

from flask import Flask

app = Flask(__name__)

@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"

app的启动，在「[flask源码阅读-开胃菜] 」中有过介绍，这里就不再赘述。我们在示例中创建了一个flask-app，然后给这个app注册了一个路由处理函数，返回一段html。跟随这个示例，我们查看flask-app的创建:

class Flask(_PackageBoundObject):
    response_class = Response
    request_class = Request
    app_ctx_globals_class = _AppCtxGlobals
    config_class = Config

    def __init__(
        self,
        import_name,
        static_url_path=None,
        static_folder="static",
        static_host=None,
        host_matching=False,
        subdomain_matching=False,
        template_folder="templates",
        instance_path=None,
        instance_relative_config=False,
        root_path=None,
    ):
        # import_name使用__name__
        _PackageBoundObject.__init__(
            self, import_name, template_folder=template_folder, root_path=root_path
        )
        ...

主要有下面三点：

_PackageBoundObject类主要实现当前app根目录获取:

class _PackageBoundObject(object):
    import_name = None

    template_folder = None

    root_path = None

    def __init__(self, import_name, template_folder=None, root_path=None):
        self.import_name = import_name
        self.template_folder = template_folder

        if root_path is None:
            root_path = get_root_path(self.import_name)

        self.root_path = root_path
        self._static_folder = None
        ...

获取根目录后，默认static中的静态文件和templates中的模板文件就可以提供读取，这样flask-app不需要额外使用werkzeug中的StaticMiddleware。比如静态文件目录的提供:

@property
def static_folder(self):
    """The absolute path to the configured static folder."""
    if self._static_folder is not None:
        return os.path.join(self.root_path, self._static_folder)

flask-app中实现wsgi-application规范:

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:
            ...
        return response(environ, start_response)
    finally:
        ...
        ctx.auto_pop(error)

def __call__(self, environ, start_response):
    return self.wsgi_app(environ, start_response)

RequestContext && AppContext

Request上下文(RequestContext)是flask中最重要的类，下面是它的构造函数:

class RequestContext(object):

    def __init__(self, app, environ, request=None, session=None):
        self.app = app
        if request is None:
            request = app.request_class(environ)
        self.request = request
        self.url_adapter = None
        try:
            self.url_adapter = app.create_url_adapter(self.request)
        except HTTPException as e:
            self.request.routing_exception = e
        self.flashes = None
        self.session = session
        self._implicit_app_ctx_stack = []
        ...

push方法中，创建app-context然后入栈，同时创建session:

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()
        # 加入request-ctx的本地栈
        self._implicit_app_ctx_stack.append(app_ctx)
    else:
        self._implicit_app_ctx_stack.append(None)

    _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.url_adapter is not None:
        self.match_request()

match_request是对request的处理，稍后介绍Request部分再详细介绍。_app_ctx_stack和_request_ctx_stack是两个个非常重要的概念，现在我们只需要记住这是2个栈即可，先后push进去app-ctx和request-ctx，以后我会单独介绍这2个概念。

RequestContext的pop方法进行出栈操作:

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

        # get rid of circular dependencies at the end of the request
        # so that we don't require the GC to be active.
        if clear_request:
            rv.request.environ["werkzeug.request"] = None

        # Get rid of the app as well if necessary.
        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 auto_pop(self, exc):
    ...
    self.pop(exc)

和入栈相反，先进行request_ctx_的出栈，再进行app_ctx的出栈。

AppContext包装App和RequestContext包装Request类似:

class AppContext(object):

    def __init__(self, app):
        self.app = app
        self.url_adapter = app.create_url_adapter(None)
        self.g = app.app_ctx_globals_class()

        # Like request context, app contexts can be pushed multiple times
        # but there a basic "refcount" is enough to track them.
        self._refcnt = 0

    def push(self):
        """Binds the app context to the current context."""
        self._refcnt += 1
        ...
        _app_ctx_stack.push(self)
        # appcontext_pushed是一个信号，在介绍blink库时候已经有过介绍
        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]
                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)

可以看到这里的push和pop与之前的RequestContext基本一致。

request && response

request解析的第1步是从match_request开始，使用url_adapter分离出http协议的url(url_rule)和query参数(view_args):

# RequestContext
def match_request(self):
    try:
        result = self.url_adapter.match(return_rule=True)
        self.request.url_rule, self.request.view_args = result
    except HTTPException as e:
        ...

第2步是使用full_dispatch_request，进行分发前的预处理:

# Flask
def full_dispatch_request(self):
    # 触发request请求的前置函数(和信号类似)
    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:
        ...
    return self.finalize_request(rv)

第3步是使用dispatch_request将request的分发请求到路由:

# Flask
def dispatch_request(self):
    # 直接从栈中获取request对象
    req = _request_ctx_stack.top.request
    ...
    rule = req.url_rule
    ...
    return self.view_functions[rule.endpoint](**req.view_args)

view_functions是通过app的router装饰器注册而来:

# Flask
@setupmethod
def add_url_rule(
    self,
    rule,
    endpoint=None,
    view_func=None,
    provide_automatic_options=None,
    **options
):
    ...
    self.url_map.add(rule)
    if view_func is not None:
        ...
        self.view_functions[endpoint] = view_func

def route(self, rule, **options):

    def decorator(f):
        endpoint = options.pop("endpoint", None)
        self.add_url_rule(rule, endpoint, f, **options)
        return f

    return decorator

request经过上述主要的3步后，业务完成，开始进入response的处理环节。先是生成response:

# Flask
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:
        ...
    return response

response处理第1步就是构建header的status:

def make_response(self, rv):
    status = headers = None
    # unpack tuple returns
    if isinstance(rv, tuple):
        len_rv = len(rv)

        # a 3-tuple is unpacked directly
        if len_rv == 3:
            rv, status, headers = rv
        # decide if a 2-tuple has status or headers
        elif len_rv == 2:
            if isinstance(rv[1], (Headers, dict, tuple, list)):
                rv, headers = rv
            else:
                rv, status = rv
    ...
    # 处理body
    rv = self.response_class(rv, status=status, headers=headers)
    status = headers = None
    ...         
    return rv

response处理第2步就是将response进行额外处理，类似request的预处理一样，比如包装上session。

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

小结

Flask实现了wsgi-applicaction的规范，对于每一个请求。Flask都会创建RequestCtx和AppCtx对象，保证在请求的生命周期中都可以很方便的获取Request对象和App对象。而对于每个请求，分成request和response两个环节，这也是我们在werkzeug中介绍过的 [洋葱模型 ] 。Flask对request的处理分:请求URL分析，request的预处理和路由派发3步；response的处理经过状态码和head/body封装，response后置处理器2步。

以上就是Flask的核心流程了，希望这道正菜大家满意，可以给一个赞或者在看。

小技巧

使用_request_ctx_stack这种全局堆栈，可以减少参数的传递. 观测下面派发request到路由的函数:

# Flask
def dispatch_request(self):
    # 直接从栈中获取request对象
    req = _request_ctx_stack.top.request
    ...
    rule = req.url_rule
    ...
    return self.view_functions[rule.endpoint](**req.view_args)

这里dispatch_request函数没有request参数，这样避免了Flask-App对request处理的每个函数都要附带上request参数。

参考链接

Copyright© 2013-2020

All Rights Reserved 京ICP备2023019179号-8