Flask一个g引发的思考 | LINSHUKAI'S BLOG

背景

最近有面试一家公司，感觉准备的不是很充分，感觉很多东西都答的挺菜的，自己写的文章里面的问题都没答上来更是汗流浃背。
所以大概列了一下其中的问题，进行了一番总结补充，重新制定一下复习的计划。
其中里面我觉得有个问题我觉得挺有意思的，我记得大概是问了一下Flask的g有使用过？具体实现的原理是啥？
我想了一下，这不就是平时用来存储一下数据的全局变量的么？有啥原理的，后面大概翻了一下Flask的源码，具体从头捋了一遍，所以就有这篇文章。

源码阅读

g

首先我们从平时导入的”from flask import g”入手，定位到flask框架代码中g，作为起点继续往下探索；

_cv_app: ContextVar[AppContext] = ContextVar("flask.app_ctx")
app_ctx: AppContext = LocalProxy(  # type: ignore[assignment]
    _cv_app, unbound_message=_no_app_msg
)
current_app: Flask = LocalProxy(  # type: ignore[assignment]
    _cv_app, "app", unbound_message=_no_app_msg
)
g: _AppCtxGlobals = LocalProxy(  # type: ignore[assignment]
    _cv_app, "g", unbound_message=_no_app_msg
)

这里面重要的就两块地方，_cv_app，g：

_cv_app这个对象实际上就是初始化一个上下文变量ContexVar，根据这个类型注解，这个上下文变量中的值是AppContext对象；
g实质就是_AppCtxGlobals对象，但是这里是通过LocalProxy代理对象进行访问；

LocalProxy.init

接下来继续分析一下LocalProxy这个代理对象是怎么代理访问_AppCtxGlobals，也就是应用上下文中的对象：（代码过长，我截取初始化中最重要的几部分）

def __init__(
        self,
        local: ContextVar[T] | Local | LocalStack[T] | t.Callable[[], T],
        name: str | None = None,
        *,
        unbound_message: str | None = None,
    ) -> None:

定义传入的参数，这里local就是_cv_app上下文变量，name就是”g”；

if name is None:
    get_name = _identity
else:
    get_name = attrgetter(name)

生成一个可调用对象，用于提取对象中的属性值。后续get_name(obj)相当于直接调用obj.name属性；

elif isinstance(local, ContextVar):

    def _get_current_object() -> T:
        try:
            obj = local.get()
        except LookupError:
            raise RuntimeError(unbound_message) from None

        return get_name(obj)

这部分有两块地方比较模糊，local.get()，get_name(obj)

由于传入local是ContexVar类型，定位到if匹配的代码，这部分是定义_get_current_object方法
local.get()其实返回的就是set进去的AppContext对象，所以obj就是AppContext类型；
get_name方法前面定义好了，所以可以理解为AppContext.g；

1	object.__setattr__(self, "_get_current_object", _get_current_object)

LocalProxy代理器绑定_get_current_object方法

LocalProxy.setattr、getattr

众所周知，当给一个类绑定属性时候，会调用类的__setattr__方法，当读取一个不存在的属性时，会调用__getattr__方法，调用一个已存在属性时，会调用__getattribute__方法；

1
2
3

__getattr__ = _ProxyLookup(getattr)
# __getattribute__ triggered through __getattr__
__setattr__ = _ProxyLookup(setattr)  # type: ignore

这部分有个地方挺有意思的，就是__getattribute__方法源代码是注释的了，并且配上通过__getattr__触发，这里可能需要绕个弯，本质上这里是代理器对象，目标对象并不是直接绑定到代理的。所以其实每次通过g去获取存储的对象是，这个代理器的__getattr__就会被触发了；
设置__setattr__方法；可以理解为后续LocalProxy的实例对象设置属性时，_ProxyLookup(setattr)(self, key, value)，这个self指的是LocalProxy的实例对象

_PorxyLookup.init

接下来，我们接到看这个_PorxyLookup对象，这个本质上就是个描述器，用于查找对象的。

描述器开始有点熟悉了，上面提到_ProxyLookup的实例当成方法一样调用的时候，应该先定位到_PorxyLookup对象__call__方法，先看看_PorxyLookup的初始化__init__：

def __init__(
    self,
    f: t.Callable | None = None,
    fallback: t.Callable | None = None,
    class_value: t.Any | None = None,
    is_attr: bool = False,
) -> None:
    bind_f: t.Callable[[LocalProxy, t.Any], t.Callable] | None

    if hasattr(f, "__get__"):
        # A Python function, can be turned into a bound method.

        def bind_f(instance: LocalProxy, obj: t.Any) -> t.Callable:
            return f.__get__(obj, type(obj))  # type: ignore

    elif f is not None:
        # A C function, use partial to bind the first argument.

        def bind_f(instance: LocalProxy, obj: t.Any) -> t.Callable:
            return partial(f, obj)

    else:
        # Use getattr, which will produce a bound method.
        bind_f = None

    self.bind_f = bind_f
    self.fallback = fallback
    self.class_value = class_value
    self.is_attr = is_attr

判断传入的f是否为描述器，很显然传入的setattr为否，且f不为None：
声明一个bind_f函数，函数用于返回partial对象，partial函数是用于固定了obj参数；
前面提到传入的是setattr，setattr的函数定义setattr(x,y,z)，正常使用需要传入三个参数，才能达到x.y = z的效果；
现在就是bind_f直接返回的是固定了第一个参数为obj的setattr函数，后续只需要调用只需要传入后两个参数即可；
(注意这个self.bind_f，后面会用到)

_PorxyLookup.call

接着继续看_PorxyLookup的__call__方法:

def __call__(self, instance: LocalProxy, *args: t.Any, **kwargs: t.Any) -> t.Any:
    """Support calling unbound methods from the class. For example,
    this happens with ``copy.copy``, which does
    ``type(x).__copy__(x)``. ``type(x)`` can't be proxied, so it
    returns the proxy type and descriptor.
    """
    return self.__get__(instance, type(instance))(*args, **kwargs)

这里其实就是当实例当做函数一样调用时会触发的__call__方法，这里可以看到instance对应着我们传入的LocalProxy实例对象，*args就是我们传入设置的参数了；
后面就是调用描述器的__get__方法了，直接传入实例对象，实例类型；

_PorxyLookup.get

这部分算我们寻找设置g这个全局上下文变量的重点，把前面做的所有事情进行一次“回收”使用：

def __get__(self, instance: LocalProxy, owner: type | None = None) -> t.Any:
    if instance is None:
        if self.class_value is not None:
            return self.class_value

        return self

    try:
        obj = instance._get_current_object()
    except RuntimeError:
        if self.fallback is None:
            raise

        fallback = self.fallback.__get__(instance, owner)

        if self.is_attr:
            # __class__ and __doc__ are attributes, not methods.
            # Call the fallback to get the value.
            return fallback()

        return fallback

    if self.bind_f is not None:
        return self.bind_f(instance, obj)

    return getattr(obj, self.name)

_get_current_object方法，前面在LocalProxy代理器对象绑定好的方法，方法里面其实最终返回的是就是一个AppContext.g对象；
self.bind_f前面初始化刚绑定好的，这里其实就是返回一个固定参数的setattr函数，setattr的第一个参数固定成obj了;
结合上面__call__方法，再把剩余需要设置的参数传进来，假设在flask框架设置g全局变量是g.name = “flask”，其实在这里执行的就是setattr(AppContext.g, “name”, “flask”);
所以就是调用目标对象AppContext中g属性的__setattr__方法；

整体总结

整体看下来就是，废了老大劲，其实就是调用AppContext的方法。而里面代码架构的核心就是代理模式，使用者就是通过g这个代理器去访问AppContext里面的g属性。而这个AppContext对象则是，存放在_cv_app这个上下文变量。而这个g实质就是在存储数据的应用上下文。

下面展示一下AppContext类：

class AppContext:
    """The app context contains application-specific information. An app
    context is created and pushed at the beginning of each request if
    one is not already active. An app context is also pushed when
    running CLI commands.
    """

    def __init__(self, app: Flask) -> None:
        self.app = app
        self.url_adapter = app.create_url_adapter(None)
        self.g: _AppCtxGlobals = app.app_ctx_globals_class()
        self._cv_tokens: list[contextvars.Token] = []

    def push(self) -> None:
        """Binds the app context to the current context."""
        self._cv_tokens.append(_cv_app.set(self))
        appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync)

这里面有趣的点就是什么时候触发这个push方法，即需要把当前的AppContext保存到_cv_app这个上下文变量中，不然你直接在flask程序启动的前调用这个g，会发现抛出一个RuntimeError的异常，因为_cv_app里面是空的；
这里不详细列了，有兴趣的再去仔细看看，答案是Flask对象中的wsgi_app方法中，当WSGI服务器调用Falsk对象作为应用程序时就会调用wsgi_app方法，wsgi_app就会推送应用程序上下文；

最后一试

上面提到很关键的代理模式，众所周知，一般来说代理模式目的就是防止调用者和执行者发生关系，所以需要一个代理对象。如果上面这么绕的代码看不懂，其实可以简洁成以下的代码，
或许你直接就豁然开朗了。

from werkzeug.local import LocalProxy
from contextvars import ContextVar


class Info:
    pass


class Person:
    def __init__(self):
        self.info = Info()


if __name__ == "__main__":
    flask_var = ContextVar("flask.context")
    p = Person()
    flask_var.set(p)

    proxy = LocalProxy(flask_var, "info", unbound_message="Error bound msg")
    proxy.name = "zhangsan"
    proxy.age = 12
    print(p.info.name, p.info.age)

所以梳理完整遍代码，好像大概知道了一点原理。。。

参考链接：

https://www.cnblogs.com/cwp-bg/p/10084480.html

https://flask.palletsprojects.com/en/3.0.x/appcontext/

背景

源码阅读

g

LocalProxy.__init__

LocalProxy.__setattr__、__getattr__

_PorxyLookup.__init__

_PorxyLookup.__call__

_PorxyLookup.__get__