12. Flask的上下文管理

Posted 2021-03-01 hq82

一、Threading.local

通过Threading.local实现本地存储隔离；当我们开启多线程来执行func函数，通过自定义的Foo类来存储数据时，我们发现最终的输出结果是全部的线程打印的都是一个最终的数字10，这是因为这样存储的数据线程之间是共享的，当最后一个线程执行func函数时，由于func函数time.sleep了1秒钟，第一个线程还没有来得及打印数据呢，就被最后一个线程传的参数给覆盖了，从而线程们打印的结果都是相同的。

from threading import local
import threading
import time

class Foo(object):
    pass

def func(num):
    Foo.num = num
    time.sleep(1)
    print(Foo.num,threading.current_thread().ident)


for i in range(1,11):
    th = threading.Thread(target=func,args=(i,))
    th.start()

我们使用Threading中的local来存储数据就可以做到数据在线程之间数据隔离。从而就可以理解为什么flask的上下文管理是基于local来实现的数据存储了。

from threading import local
import threading
import time

foo = local()

def func(num):
    foo.num = num
    time.sleep(1)
    print(foo.num,threading.current_thread().ident)


for i in range(1,11):
    th = threading.Thread(target=func,args=(i,))
    th.start()

二、实例化一个Flask应用

使用app = Flask(__name__)，可以实例化一个Flask应用。实例化的Flask应用有一些要点或特性需要注意一下：

1. 对于请求和响应的处理，Flask使用werkzeug库中的Request类和Response类。
2. 对于URL模式的处理，Flask应用使用werkzeug库中的Map类和Rule类，每一个URL模式对应一个Rule实例，这些Rule实例最终会作为参数传递给Map类构造包含所有URL模式的一个“地图”。这个地图可以用来匹配请求中的URL信息。
3. 当实例化一个Flask应用app（这个应用的名字可以随便定义）之后，对于如何添加URL模式，Flask采取了一种更加优雅的模式，对于这点可以和Django的做法进行比较。Flask采取装饰器的方法，将URL规则和视图函数结合在一起写，其中主要的函数是route。这样写视图函数，会将‘/‘这条URL规则和视图函数index()联系起来，并且会形成一个Rule实例，再添加进Map实例中去。当访问‘/‘时，会执行index()。
4. 实例化Flask应用时，会创造一个Jinja环境，这是Flask自带的一种模板引擎。
5. 实例化的Flask应用是一个可调用对象。在前面讲到，Web应用要遵循WSGI规范，就要实现一个函数或者一个可调用对象webapp(environ, start_response)，以方便服务器或网关调用。Flask应用通过__call__(environ, start_response)方法可以让它被服务器或网关调用。
   注意到调用该方法会执行wsgi_app(environ, start_response)方法，之所以这样设计是为了在应用正式处理请求之前，可以加载一些“中间件”,以此改变Flask应用的相关特性。
6. Flask应用还有一些其他的属性或方法，用于整个请求和响应过程。

三、调用Flask应用

上面部分分析了实例化的Flask应用长什么样子。当一个完整的Flask应用实例化后，可以通过调用app.run()方法运行这个应用。

Flask应用的run()方法会调用werkzeug.serving模块中的run_simple方法。这个方法会创建一个本地的测试服务器，并且在这个服务器中运行Flask应用。

当服务器开始调用Flask应用后，便会触发Flask应用的_call_(environ, start_response)方法。其中environ由服务器产生，start_response在服务器中定义。

上面我们分析到当Flask应用被调用时会执行wsgi_app(environ, start_response)方法。可以看出，wsgi_app是真正被调用的WSGI应用，之所以这样设计，就是为了在应用正式处理请求之前，wsgi_app可以被一些“中间件”装饰，以便先行处理一些操作。

四、请求和应用上下文

在 flask 中，视图函数需要知道它执行情况的请求信息（请求的 url，参数，方法等）以及应用信息（应用中初始化的数据库等），才能够正确运行。

最直观地做法是把这些信息封装成一个对象，作为参数传递给视图函数。但是这样的话，所有的视图函数都需要添加对应的参数，即使该函数内部并没有使用到它。

flask 的做法是把这些信息作为类似全局变量的东西，视图函数需要的时候，可以使用 from flask import request 获取。但是这些对象和全局变量不同的是——它们必须是动态的，因为在多线程或者多协程的情况下，每个线程或者协程获取的都是自己独特的对象，不会互相干扰。

那么如何实现这种效果呢？如果对 python 多线程比较熟悉的话，应该知道多线程中有个非常类似的概念 threading.local，可以实现多线程访问某个变量的时候只看到自己的数据。内部的原理说起来也很简单，这个对象有一个字典，保存了线程 id 对应的数据，读取该对象的时候，它动态地查询当前线程 id 对应的数据。

flask 中有两种上下文：application context 和 request context。上下文有关的内容定义在 globals.py 文件。

from functools import partial
from werkzeug.local import LocalStack, LocalProxy


_request_ctx_err_msg = ‘‘‘Working outside of request context.

This typically means that you attempted to use functionality that needed
an active HTTP request.  Consult the documentation on testing for
information about how to avoid this problem.‘‘‘
_app_ctx_err_msg = ‘‘‘Working outside of application context.

This typically means that you attempted to use functionality that needed
to interface with the current application object in some way. To solve
this, set up an application context with app.app_context().  See the
documentation for more information.‘‘‘
def _lookup_req_object(name):
    top = _request_ctx_stack.top
    if top is None:
        raise RuntimeError(_request_ctx_err_msg)
    return getattr(top, name)

def _lookup_app_object(name):
    top = _app_ctx_stack.top
    if top is None:
        raise RuntimeError(_app_ctx_err_msg)
    return getattr(top, name)

def _find_app():
    top = _app_ctx_stack.top
    if top is None:
        raise RuntimeError(_app_ctx_err_msg)
    return top.app

# context locals
_request_ctx_stack = LocalStack()  #  {"__storage__":{ },"__ident_func__" : get_ident}
_app_ctx_stack = LocalStack()
current_app = LocalProxy(_find_app)
request = LocalProxy(partial(_lookup_req_object, ‘request‘))
session = LocalProxy(partial(_lookup_req_object, ‘session‘))
g = LocalProxy(partial(_lookup_app_object, ‘g‘))

flask 提供两种上下文：application context 和 request context 。application context 又演化出来两个变量 current_app 和 g，而 request context 则演化出来 request 和 session。

这里的实现用到了两个东西：LocalStack 和 LocalProxy。它们两个的结果就是我们可以动态地获取两个上下文的内容，在并发程序中每个视图函数都会看到属于自己的上下文，而不会出现混乱。

LocalStack 和 LocalProxy 都是 werkzeug 提供的，定义在 local.py 文件中。在分析这两个类之前，我们先介绍这个文件另外一个基础的类 Local。Local 就是实现了类似 threading.local 的效果——多线程或者多协程情况下全局变量的隔离效果。下面是它的代码：

import copy
from functools import update_wrapper
from werkzeug.wsgi import ClosingIterator
from werkzeug._compat import PY2, implements_bool

try:
    from greenlet import getcurrent as get_ident
except ImportError:
    try:
        from thread import get_ident
    except ImportError:
        from _thread import get_ident

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

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

    def __iter__(self):
        return iter(self.__storage__.items())

    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":{"key1":"value1","key2":"value2",...}}。最外面字典 key 是线程或者协程的 identity，value 是另外一个字典，这个内部字典就是用户自定义的 key-value 键值对。

用户访问实例的属性，就变成了访问内部的字典，外面字典的 key 是自动关联的。self.ident_func() 就是执行get_ident()，得到的是协程的 get_current 或者线程的 get_ident，从而获取当前代码所在线程或者协程的 id。

除了这些基本操作之外，Local 还实现了_release_local_ ，用来清空（析构）当前线程或者协程的数据（状态）。call 操作来创建一个 LocalProxy 对象，LocalProxy 会在下面讲到。

理解了 Local，我们继续回来看另外两个类。

LocalStack 是基于 Local 实现的栈结构。如果说 Local 提供了多线程或者多协程隔离的属性访问，那么 LocalStack 就提供了隔离的栈访问。下面是它的实现代码，可以看到它提供了 push、pop 和 top 方法。

release_local 可以用来清空当前线程或者协程的栈数据，call 方法返回当前线程或者协程栈顶元素的代理对象。

class LocalStack(object):

    def __init__(self):
        self._local = Local()  //  {"__storage__":{},"__ident_func__":get_ident}

    def __release_local__(self):
        self._local.__release_local__()

    def __call__(self):
        def _lookup():
            rv = self.top
            if rv is None:
                raise RuntimeError(‘object unbound‘)
            return rv
        return LocalProxy(_lookup)


    //  push、pop 和 top 三个方法实现了栈的操作，
    //  可以看到栈的数据是保存在 self._local.stack 属性中的
    def push(self, obj):
        """Pushes a new item to the stack"""
        rv = getattr(self._local, ‘stack‘, None)  //  {"__storage__":{},"__ident_func__":get_ident}
        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):
        try:
            return self._local.stack[-1]
        except (AttributeError, IndexError):
            return None

我们在之前看到了 request context 的定义，它就是一个 LocalStack 的实例：

_request_ctx_stack = LocalStack()   // {"__storage__":{},"__ident_func__":get_ident}

它会把当前线程或者协程的请求都保存在栈里，等使用的时候再从里面读取。

LocalProxy 是一个 Local 对象的代理，负责把所有对自己的操作转发给内部的 Local 对象。LocalProxy 的构造函数介绍一个 callable 的参数，这个 callable 调用之后需要返回一个 Local 实例，后续所有的属性操作都会转发给 callable 返回的对象。

@implements_bool
class LocalProxy(object):

    __slots__ = (‘__local‘, ‘__dict__‘, ‘__name__‘, ‘__wrapped__‘)

    def __init__(self, local, name=None):
        object.__setattr__(self, ‘_LocalProxy__local‘, local)
        object.__setattr__(self, ‘__name__‘, name)
        if callable(local) and not hasattr(local, ‘__release_local__‘):
            # "local" is a callable that is not an instance of Local or
            # LocalManager: mark it as a wrapped function.
            object.__setattr__(self, ‘__wrapped__‘, local)

    def _get_current_object(self):
        """Return the current object"""
        if not hasattr(self.__local, ‘__release_local__‘):
            return self.__local()
        try:
            return getattr(self.__local, self.__name__)
        except AttributeError:
            raise RuntimeError(‘no object bound to %s‘ % self.__name__)

    @property
    def __dict__(self):
        try:
            return self._get_current_object().__dict__
        except RuntimeError:
            raise AttributeError(‘__dict__‘)

    def __getattr__(self, name):
        if name == ‘__members__‘:
            return dir(self._get_current_object())
        return getattr(self._get_current_object(), name)

    def __setitem__(self, key, value):
        self._get_current_object()[key] = value

    def __delitem__(self, key):
        del self._get_current_object()[key]

这里实现的关键是把通过参数传递进来的 Local 实例保存在 __local 属性中，并定义了 _get_current_object() 方法获取当前线程或者协程对应的对象。

看下面这段代码就能看明白，由于_request_ctx_stack 是多线程或者协程隔离的栈结构，request 每次都会调用 _lookup_req_object 栈头部的数据来获取保存在里面的 requst context。

_request_ctx_stack = LocalStack()  // {"__storage__":{},"__ident_func__":get_ident}
request = LocalProxy(partial(_lookup_req_object, ‘request‘))
session = LocalProxy(partial(_lookup_req_object, ‘session‘))

那么请求上下文信息是什么时候被放在 stack 中呢？还记得之前介绍的 wsgi_app() 方法有下面代码吗？

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


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)

每次在调用 app.call 的时候，都会把对应的请求信息压栈，最后执行完请求的处理之后把它出栈。

我们来看看request_context， 这个 方法只有一行代码：

def request_context(self, environ):
    return RequestContext(self, environ)

它调用了 RequestContext，并把 self 和请求信息的字典 environ 当做参数传递进去。追踪到 RequestContext 定义的地方，它出现在 ctx.py 文件中，代码如下：

class RequestContext(object):

    def __init__(self, app, environ, request=None):
        self.app = app
        if request is None:
            request = app.request_class(environ)
        self.request = request
        self.url_adapter = app.create_url_adapter(self.request)
        self.flashes = None
        self.session = None

    def match_request(self):
        """Can be overridden by a subclass to hook into the matching
        of the request.
        """
        try:
            url_rule, self.request.view_args =                 self.url_adapter.match(return_rule=True)
            self.request.url_rule = url_rule
        except HTTPException as e:
            self.request.routing_exception = e

    def push(self):
        top = _request_ctx_stack.top # {"__storage__":{},"__ident_func__":get_ident}
        if top is not None and top.preserved:
            top.pop(top._preserved_exc)

        # Before we push the request context we have to ensure that there
        # is an application context.
        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)

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

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

    def __exit__(self, exc_type, exc_value, tb):
        self.auto_pop(exc_value)

        if BROKEN_PYPY_CTXMGR_EXIT and exc_type is not None:
            reraise(exc_type, exc_value, tb)

    def __repr__(self):
        return ‘<%s ‘%s‘ [%s] of %s>‘ % (
            self.__class__.__name__,
            self.request.url,
            self.request.method,
            self.app.name,
        )

每个 request context 都保存了当前请求的信息，比如 request 对象和 app 对象。在初始化的最后，还调用了 match_request 实现了路由的匹配逻辑。

push 操作就是把该请求的 ApplicationContext（如果 _app_ctx_stack 栈顶不是当前请求所在 app ，需要创建新的 app context） 和 RequestContext 有关的信息保存到对应的栈上，压栈后还会保存 session 的信息； pop 则相反，把 request context 和 application context 出栈，做一些清理性的工作。

到这里，上下文的实现就比较清晰了：每次有请求过来的时候，flask 会先创建当前线程或者进程需要处理的两个重要上下文对象，把它们保存到隔离的栈里面，这样视图函数进行处理的时候就能直接从栈上获取这些信息。