flask源码分析

2023/03/01 python flask 共 8734 字,约 25 分钟

flask工作原理以及源码分析,本文所有的源码基于flask v0.1版本

概念

单说Flask的话,代码量确实很少,但是Flask完全建立在Werkzeug之上,如果把Werkzeug的代码加起来,代码量 可就不少了。。。 但这里主要分析flask的原理,其中有比较好设计,Werkzeug代码暂时不分析了。。

这里有flask的几个基本概念,简单说明一下

  • Werkzeug

Werkzeug是一个WSGI工具包,它可以作为web框架的底层库。负责核心的逻辑模块,比如路由、请求和应答的封装、WSGI 相关的函数等;

  • jinja

负责模板的渲染,主要用来渲染返回给用户的 html 文件内容。

Flask是一个基于Python开发并且依赖jinja2模板和Werkzeug WSGI服务的一个微型框架,对于Werkzeug,它只是工具包,其用于接收http请求并对请求进行预处理, 然后触发Flask框架;对于jinja2模板,它用来实现对模板的处理,将模板和数据进行渲染,将渲染后的字符串返回给用户浏览器。

解析源码

run

这里主要解析flask核心源码及实现方式,jinja和Werkzeug暂时跳过。。。

首先一个最简单的demo代码为

from flask import Flask
app = Flask(__name__)

@app.route("/")
def hello():
    return "Hello World!"

if __name__ == "__main__":
    app.run()

通过Debug调试工作可以整理其调用栈如下:

app.run()
    run_simple(host, port, self, **options)
      __call__(self, environ, start_response)
        wsgi_app(self, environ, start_response)

run_simple 为引用的wsgi方法,实际调用的__call__方法

把Flask类简化一下代码

class Flask:
    def __init__(self, package_name):

        self.package_name = package_name 
        self.root_path = _get_package_path(self.package_name)

        self.view_functions = {}
        self.error_handlers = {}
        self.before_request_funcs = []
        self.after_request_funcs = []
        self.url_map = Map()
        
    def run(self, host='localhost', port=5000, **options):
        """Runs the application on a local development server.  If the
        :attr:`debug` flag is set the server will automatically reload
        for code changes and show a debugger in case an exception happened.
    
        :param host: the hostname to listen on.  set this to ``'0.0.0.0'``
                     to have the server available externally as well.
        :param port: the port of the webserver
        :param options: the options to be forwarded to the underlying
                        Werkzeug server.  See :func:`werkzeug.run_simple`
                        for more information.
        """
        from werkzeug import run_simple
        if 'debug' in options:
            self.debug = options.pop('debug')
        options.setdefault('use_reloader', self.debug)
        options.setdefault('use_debugger', self.debug)
        return run_simple(host, port, self, **options)
    
    def wsgi_app(self, environ, start_response):
        """The actual WSGI application.  This is not implemented in
        `__call__` so that middlewares can be applied:

            app.wsgi_app = MyMiddleware(app.wsgi_app)

        :param environ: a WSGI environment
        :param start_response: a callable accepting a status code,
                               a list of headers and an optional
                               exception context to start the response
        """
        with self.request_context(environ):
            rv = self.preprocess_request()
            if rv is None:
                rv = self.dispatch_request()
            response = self.make_response(rv)
            response = self.process_response(response)
            return response(environ, start_response)
    
    def __call__(self, environ, start_response):
        """Shortcut for :attr:`wsgi_app`"""
        return self.wsgi_app(environ, start_response)

这下可能稍微清晰一点了,接下来每一步部分分开说明。

view_functions中保存了视图函数(处理用户请求的函数,如上面的hello()),

error_handlers 这个字典用来保存所有的错误处理视图函数,字典的 key 是错误类型码

before_request_funcs 这个列表用来保存在请求被分派之前应当执行的函数

before_first_request_funcs 在接收到第一个请求的时候应当执行的函数。

after_request_funcs 这个列表中的函数在请求完成之后被调用,响应对象会被传给这些函数

self.url_map 这里设置了一个 url_map 属性,并把它设置为一个 Map 对象,用以保存URI到视图函数的映射,即保存app.route()这个装饰器的信息

其中最核心的就是 wsgi_app方法,他有一个上下文 request_context(environ),看一下具体实现

class _RequestContext(object):
    """The request context contains all request relevant information.  It is
    created at the beginning of the request and pushed to the
    `_request_ctx_stack` and removed at the end of it.  It will create the
    URL adapter and request object for the WSGI environment provided.
    """

    def __init__(self, app, environ):
        self.app = app
        self.url_adapter = app.url_map.bind_to_environ(environ)
        self.request = app.request_class(environ)
        self.session = app.open_session(self.request)
        self.g = _RequestGlobals()
        self.flashes = None

    def __enter__(self):
        _request_ctx_stack.push(self)

    def __exit__(self, exc_type, exc_value, tb):
        # do not pop the request stack if we are in debug mode and an
        # exception happened.  This will allow the debugger to still
        # access the request object in the interactive shell.
        if tb is None or not self.app.debug:
            _request_ctx_stack.pop()
            
def request_context(self, environ):
    """Creates a request context from the given environment and binds
    it to the current context.  This must be used in combination with
    the `with` statement because the request is only bound to the
    current context for the duration of the `with` block.

    Example usage::

        with app.request_context(environ):
            do_something_with(request)

    :params environ: a WSGI environment
    """
    return _RequestContext(self, environ)

可以看到初始化实例变量,并在 __enter__ 方法中将实例加入 context locals _request_ctx_stack__exit__方法取出实例对象

接着 preprocess_request 我们常见的钩子函数,在请求之前做处理,依次执行 self.before_request_funcs中的方法

def preprocess_request(self):
    """Called before the actual request dispatching and will
    call every as :meth:`before_request` decorated function.
    If any of these function returns a value it's handled as
    if it was the return value from the view and further
    request handling is stopped.
    """
    for func in self.before_request_funcs:
        rv = func()
        if rv is not None:
            return rv

大家都知道before_request这里用了一个装饰器的设计,把将要处理的函数用装饰器添加到 self.before_request_funcs(0.1版本还没有这个设计,直接用函数注册到实例中了。。问题不大不影响主要逻辑)

def before_request(self, f):
    """Registers a function to run before each request."""
    self.before_request_funcs.append(f)
    return f

接下来就 dispatch_request() 函数的处理了

def dispatch_request(self):
    """Does the request dispatching.  Matches the URL and returns the
    return value of the view or error handler.  This does not have to
    be a response object.  In order to convert the return value to a
    proper response object, call :func:`make_response`.
    """
    try:
        endpoint, values = self.match_request()
        return self.view_functions[endpoint](**values)
    except HTTPException as e:
        handler = self.error_handlers.get(e.code)
        if handler is None:
            return e
        return handler(e)
    except Exception as e:
        handler = self.error_handlers.get(500)
        if self.debug or handler is None:
            raise
        return handler(e)

match_request() 获取到函数名endpoint和参数values,然后根据endpoint从view_functions获取处理函数并调用。

def match_request(self):
    """Matches the current request against the URL map and also
    stores the endpoint and view arguments on the request object
    is successful, otherwise the exception is stored.
    """
    rv = _request_ctx_stack.top.url_adapter.match()
    request.endpoint, request.view_args = rv
    return rv

路由

那么到了这里说明路由是怎么建立绑定的,我们来看看app.route()方法中做了什么:

def route(self, rule, **options):
    def decorator(f):
        self.add_url_rule(rule, f.__name__, **options)
        self.view_functions[f.__name__] = f
        return f
    return decorator
    
def add_url_rule(self, rule, endpoint, **options):
    options['endpoint'] = endpoint
    options.setdefault('methods', ('GET',))
    self.url_map.add(Rule(rule, **options))

可以看到吗,他直接用装饰器将路由和对应的函数加入到 self.view_functions

上下文

这里还有一个比较重要的设计,包括(应用上下文application context 和 请求上下文request context)

在 Flask 中,上下文是一种用于在不同部分之间传递信息的机制。Flask 使用上下文来确保在同一个请求或应用程序上下文中共享数据,这对于处理请求和响应、访问全局对象等非常有用。

视图函数需要知道它执行情况的请求信息(请求的 url,参数,方法等)以及应用信息(应用中初始化的数据库等),才能够正确运行。把这些信息作为类似全局变量的东西,视图函数需要的时候,可以使用 from flask import request 获取。

应用上下文(App Context):

应用上下文是全局的,它在整个应用程序生命周期中都存在。 它主要用于存储应用程序全局对象,如数据库连接、配置设置等。 使用 app.app_context() 可以获取应用上下文。 请求上下文(Request Context):

请求上下文是与每个请求相关的,每次请求进来时都会创建一个新的请求上下文。 它主要用于存储与请求相关的数据,如请求参数、会话数据等。 使用 flask.request 可以获取请求上下文。

application context 演化出来两个变量 current_appg;

request context 则演化出来 requestsession

# context locals
_request_ctx_stack = LocalStack()
current_app = LocalProxy(lambda: _request_ctx_stack.top.app)
request = LocalProxy(lambda: _request_ctx_stack.top.request)
session = LocalProxy(lambda: _request_ctx_stack.top.session)
g = LocalProxy(lambda: _request_ctx_stack.top.g)

那么这些全局变量是否可以在多线程环境下调用,肯定可以的。 由变量名可知,_request_ctx_stack是一个栈,跟进LocalStack类分析源码

class LocalStack(object):

    def __init__(self):
        self._local = Local()

    def push(self, obj):
        rv = getattr(self._local, 'stack', None)
        if rv is None:
            self._local.stack = rv = []
        rv.append(obj)
        return rv

    def pop(self):
        stack = getattr(self._local, 'stack', None)
        if stack is None:
            return None
        elif len(stack) == 1:
            release_local(self._local)
            return stack[-1]
        else:
            return stack.pop()
            
    @property
    def top(self):
        """The topmost item on the stack.  If the stack is empty,
        `None` is returned.
        """
        try:
            return self._local.stack[-1]
        except (AttributeError, IndexError):
            return None

class Local(object):
    __slots__ = ('__storage__', '__ident_func__')

    def __init__(self):
        # 数据保存在 __storage__ 中,后续访问都是对该属性的操作
        object.__setattr__(self, '__storage__', {})
        object.__setattr__(self, '__ident_func__', get_ident)

    def __call__(self, proxy):
        """Create a proxy for a name."""
        return LocalProxy(self, proxy)

    # 清空当前线程/协程保存的所有数据
    def __release_local__(self):
        self.__storage__.pop(self.__ident_func__(), None)

    # 下面三个方法实现了属性的访问、设置和删除。
    # 注意到,内部都调用 `self.__ident_func__` 获取当前线程或者协程的 id,然后再访问对应的内部字典。
    # 如果访问或者删除的属性不存在,会抛出 AttributeError。
    # 这样,外部用户看到的就是它在访问实例的属性,完全不知道字典或者多线程/协程切换的实现
    def __getattr__(self, name):
        try:
            return self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)

    def __setattr__(self, name, value):
        ident = self.__ident_func__()
        storage = self.__storage__
        try:
            storage[ident][name] = value
        except KeyError:
            storage[ident] = {name: value}

    def __delattr__(self, name):
        try:
            del self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)

Local类有两个成员变量,分别是__storage__和__ident_func__

__storage__是一个字典,最外面字典 key 是线程或者协程的 identity,value 是另外一个字典,这个内部字典就是用户自定义的 key-value 键值对。用户访问实例的属性,就变成了访问内部的字典,外面字典的 key 是自动关联的。

__ident_func__是一个函数。这个函数的含义是,获取当前线程的id(或协程的id)。

Local类还自定义了__getattr__和__setattr__这两个方法,也就是说,我们在操作self.local.stack时, 会调用__setattr__和__getattr__方法。

LocalProxy 是一个 Local 对象的代理,负责把所有对自己的操作转发给内部的 Local对象。

通过LocalStack和LocalProxy这样的Python魔法,每个线程访问当前请求中的数据(request, session)时, 都好像都在访问一个全局变量,但是,互相之间又互不影响。

总结

到了这里主要的流程已经过了一边了,后面还有 响应处理process_response,异常处理error_handlers没有分析,但实现都差不多,相对来说这一版flask的源码非常简单,但重要的能 从中领悟到开发者的设计思想,这对于我们才是最重要的。

文档信息

搜索

    内容