前言

在前面中，我分享过关于定义一个全局上下文request的方法，最初的设想都是通过我们的中间件来初始化我们的全局的request。后来觉得基于中间件处理相关上下文的时候，多少还是有点性能的问题，后知后觉之后，其实也觉得通过依赖注入实习也更简便，于是乎又了下面的初步的改变。

首先我们定义我们的全局的request上下文的对象：

import contextlib
from contextvars import ContextVar
from .bind_ import bind_contextvar
from starlette.requests import Request
from fastapi import Request, Response, BackgroundTasks

# 定义请求上下文的ContextVar变量
request_var: ContextVar[Request] = ContextVar("request")
# 绑定请求上下文的ContextVar变量
context_request: Request = bind_contextvar(request_var)

class GlobalContextRequest:
    """
    全局请求上下文类

    Attributes:
        name (str): 事件名称
    """

    def __init__(self, *, name: str = 'event-name'):
        self.name = name

    async def __call__(self, request: Request,background_tasks: BackgroundTasks):
        """
        请求上下文管理器

        Args:
            request (Request): 请求对象

        Returns:
            None
        """
        pass
        # 设置请求上下文
        request.state.current_background_tasks = background_tasks
        token = request_var.set(request)
        try:
            yield
        finally:
            # 清空当前的请求
            request_var.reset(token)

在上面的代码中我们通过是context_request: Request = bind_contextvar(request_var)来定义我们全局的上下问，后续我们通过context_request就可以实现类似我们的Flask中那个request,这样我们在有需要的使用到的地方，就不需要显示定义透传我们的request对象到上下文中。 其中我们的GlobalContextRequest的一个类类型的依赖性项，它异步方法__call__，其中在该函数内部，我们，也注入了其他依赖性，比如：

• request: Request
• background_tasks: BackgroundTasks 上面这两项我们都可以理解是两个依赖性。

在__call__方法中，首先设置请求对象的state.current_background_tasks属性为传入的BackgroundTasks对象。 使用request_var.set(request)保存当前请求对象的引用，也就是开始初始化我们的全局请求下文的对象，且最终在finally块中会对它进行重置token的操作，也就是将ContextVar变量request_var恢复到它的原始状态，即清除之前保存的请求对象引用。

当初始化完成相关的全局变量后，接着我们使用yield关键字允许请求上下文管理器内的代码暂停执行，这样我们可以中途暂停次操作，待结束的时候会进行执行只管执行处理。 我们再看看我们的绑定上下文方法：

def bind_contextvar(contextvar):
    class ContextVarBind:
        """
        绑定上下文变量的类

        该类用于绑定上下文变量，通过该类可以访问上下文变量中的属性、索引操作等

        Args:
            contextvar: 上下文变量对象

        Attributes:
            None

        Returns:
            ContextVarBind: 上下文变量绑定对象

        """

        __slots__ = ()

        def __getattr__(self, name):
            """
            动态属性获取方法

            通过该方法获取上下文变量对象的属性

            Args:
                name (str): 属性名

            Returns:
                object: 属性值

            Raises:
                AttributeError: 如果属性不存在

            """

            return getattr(contextvar.get(), name)

        def __setattr__(self, name, value):
            """
            动态属性设置方法

            通过该方法设置上下文变量对象的属性

            Args:
                name (str): 属性名
                value: 属性值

            Returns:
                None

            Raises:
                AttributeError: 如果属性不存在

            """

            setattr(contextvar.get(), name, value)

        def __delattr__(self, name):
            """
            动态属性删除方法

            通过该方法删除上下文变量对象的属性

            Args:
                name (str): 属性名

            Returns:
                None

            Raises:
                AttributeError: 如果属性不存在

            """

            delattr(contextvar.get(), name)

        def __getitem__(self, index):
            """
            索引获取方法

            通过该方法获取上下文变量对象的指定索引位置的值

            Args:
                index: 索引

            Returns:
                object: 索引位置的值

            Raises:
                None

            """

            return contextvar.get()[index]

        def __setitem__(self, index, value):
            """
            索引设置方法

            通过该方法设置上下文变量对象的指定索引位置的值

            Args:
                index: 索引
                value: 值

            Returns:
                None

            Raises:
                None

            """

            contextvar.get()[index] = value

        def __delitem__(self, index):
            """
            索引删除方法

            通过该方法删除上下文变量对象的指定索引位置的值

            Args:
                index: 索引

            Returns:
                None

            Raises:
                None

            """

            del contextvar.get()[index]

    return ContextVarBind()

上面的绑定方法主要是让我们后续导入了context_request可以直接的访问他包含对应的属性或方法。儿不需要在通过类似的contextvar.get()之后再进行相关属性或方法的访问。

而使用此依赖性也非常的简单，我们只需要在我们全局app注入即可，或者在你需要使用到此类相关功能的地方注入即可。如下使用方式所示：

需注意既然我们的有了全局了request，同样的我也可以有全局的BackgroundTasks，就是我们上面多注入了关于我们的background_tasks: BackgroundTasks的原因，如此我们就也随意在需要把任务加到后台的地方直接的使用：

context_request.state.current_background_tasks.addtask的方法添加后台任务！

但是上面我个人感觉使用起来还是有点局限，也就是说我们整个的封装仅仅是围绕一个类似本地作用域的全局上下文，假如我们需要更多的全局或本地的相关上下文的话，此时我们应该怎么处理呢？

其实我们可以再进一步进行封装，如下的代码示例所示：

from contextlib import contextmanager
from contextvars import ContextVar
import inspect
if hasattr(inspect, 'Parameter') and hasattr(inspect.Parameter, 'empty'):
    # 使用新版本的写法（Python 3.10+）
    empty_value = inspect.Parameter.empty
else:
    # 使用旧版本的写法
    empty_value = inspect._empty

from typing import Any, ClassVar, Dict, Iterator, Mapping, Optional, Callable, Annotated
from types import MappingProxyType


class Singleton:
    """一个用于实现单例设计模式的类。"""
    _instance: ClassVar[Optional[Any]] = None

    def __new__(cls, *args: Any, **kwargs: Any) -> Any:
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    @classmethod
    def _drop(cls) -> None:
        cls._instance = None


class FastAppContextRepo(Singleton):
    """一个用于表示上下文存储器的类。"""
    _global_context: Dict[str, Any]  # 全局上下文字典，存储键值对数据,类似全局变量
    _scope_context: Dict[str, ContextVar[Any]]  # 范围上下文字典，存储键值对数据，其中值为ContextVar对象

    def __init__(self) -> None:
        # 创建一个字典，键为"context"，值为self
        self._global_context = {"context": self}
        # 创建一个空字典
        self._scope_context = {}
        # 使用self._global_context创建一个MappingProxyType对象，并将其赋值给self._read_only_global_context
        # 你想要保护一个字典，防止它被修改，或者当你想要创建一个只读的字典视图时。
        # 例如，你可以使用 MappingProxyType 来包装一个字典，这样对这个字典的所有修改都会被忽略，从而保护里面的数据。
        self._read_only_global_context = MappingProxyType(self._global_context)

    def set_global(self, key: str, v: Any) -> None:
    """设置全局上下文的值。"""
    if  key not in self._global_context:
        self._global_context[key] = v

    def reset_global(self, key: str) -> None:
        """重置全局上下文的值。"""
        self._global_context.pop(key, None)

    def set_local(self, key: str, value: Any) -> "ContextVar[Any].Token":
        """设置局部上下文的值。"""
        context_var = self._scope_context.get(key)
        if context_var is None:
            context_var = ContextVar(key, default=None)
            self._scope_context[key] = context_var
        return context_var.set(value)

    def reset_local(self, key: str, token: "ContextVar[Any].Token") -> None:
        """重置局部上下文的值。"""
        self._scope_context[key].reset(token)

    def get_local(self, key: str, default: Any = None) -> Any:
        """获取局部上下文的值。"""
        context_var = self._scope_context.get(key)
        return context_var.get() if context_var is not None else default

    def clear(self) -> None:
        """清空上下文存储器。"""
        self._global_context = {"context": self}
        self._scope_context.clear()

    def get(self, key: str, default: Any = None) -> Any:
        """获取全局上下文的值。,如果没有则获取局部的"""
        return self._global_context.get(key, self.get_local(key, default))

    def resolve(self, argument: str) -> Any:
        """解析复杂键。
        这个函数用于解析一个复杂键，即键中包含点号分隔的多个部分。首先根据点号分割键字符串，获取第一个部分first，
        然后通过self.get方法获取对应的值v。如果v为None，则抛出AttributeError异常。
        接着遍历剩余的部分列表keys，依次获取v的对应值。
        如果在获取值的过程中出现KeyError或AttributeError异常，则抛出AttributeError异常。最后返回解析得到的值v。
        Args:
            argument (str): 复杂键字符串。

        Returns:
            Any: 解析得到的值。

        Raises:
            AttributeError: 当self.get(first, None)返回None时，表示`self.context`不包含`first`键。
            AttributeError: 当v为None时，尝试访问v[i]或getattr(v, i)会引发KeyError或AttributeError。

        """
        first, *keys = argument.split(".")

        v = self.get(first, None)
        if v is None:
            raise AttributeError(f"`{self.context}` does not contain `{first}` key")

        for i in keys:
            try:
                v = v[i] if isinstance(v, Mapping) else getattr(v, i)
            except (KeyError, AttributeError):
                raise AttributeError(f"Attribute `{i}` not found in context")
        return v

    def resolve_context_by_name(self,
            name: str,
            default: Any,
    ) -> Any:
        value: Any = empty_value

        try:
            value = self.resolve(name)

        except (KeyError, AttributeError):
            if default is not empty_value:
                value = default
                self.set_global(name, value)
        return value

    def __getattr__(self, __name: str) -> Any:
        """动态获取上下文的值。"""
        return self.get(__name)

    @property
    def context(self) -> Mapping[str, Any]:
        """获取上下文映射。
        这个函数返回一个上下文映射，它包含全局上下文和作用域上下文。全局上下文是一个只读的字典，作用域上下文是一个字典，其中的值是可调用的对象。
        函数使用字典解包来合并这两个上下文，形成最终的上下文映射。
        """
        return {**self._read_only_global_context, **{i: j.get() for i, j in self._scope_context.items()}}

    @contextmanager
    def scope(self, key: str, value: Any) -> Iterator[None]:
        """创建一个上下文作用域。"""
        token = self.set_local(key, value)
        try:
            yield
        finally:
            self.reset_local(key, token)


from typing import cast
# cast 函数在这里的作用是将 FastAppContextRepo() 的实例强制转换为 FastAppContextRepo 类型。
# 这样做是为了解决静态类型检查器无法推断 FastAppContextRepo() 的类型的问题。
# 使用 cast 函数可以显式地告诉类型检查器我们希望将其视为 FastAppContextRepo 类型。
appContextRepo: FastAppContextRepo = cast(FastAppContextRepo, FastAppContextRepo())  # type: ignore[redundant-cast]

紧接着我们可以开始进行相关局部上下文变量或全局上下文变量的设置处理，比如我们在FastAPI中可以这样继续使用：

使用示例

import contextlib
import time
from contextlib import asynccontextmanager

import aiojobs
from fastapi import FastAPI, Request,BackgroundTasks,Depends
import asyncio


from KKKK import appContextRepo

class GlobalContextRequest:


    async def __call__(self,request: Request, background_tasks: BackgroundTasks) -> None:
       appContextRepo.set_global('app', request.app)
       appContextRepo.set_global('scheduler', aiojobs.Scheduler())
       # 设置局部
       request.state.background_tasks = background_tasks
       token_request = appContextRepo.set_local('request', request)
       token_background_tasks = appContextRepo.set_local('background_tasks', background_tasks)

       try:
           yield
       except Exception as e:
           # 异常处理逻辑
           ...
           pass
       finally:
           appContextRepo.reset_local('request', token_request)
           appContextRepo.reset_local('background_tasks', token_background_tasks)


def write_notification(email: str, message=""):
    print("邮件准备中····")
    time.sleep(10)
    print("邮件已发送····")

async def coro(timeout):
    """
    异步函数：等待一段时间

    参数：
    - timeout (float): 等待的时长
    返回值：
    无
    """
    await asyncio.sleep(timeout)
    print("输出！！！")



app = FastAPI(dependencies=[Depends(GlobalContextRequest())])


@app.get("/")
async def send_notification():
    appContextRepo.background_tasks.add_task(write_notification, '308711822@qq.com', message="some notification")
    for i in range(100):
        # 生成任务
        await appContextRepo.scheduler.spawn(coro(5))
    return {"message": "Notification sent in the background"}




# 运行 FastAPI 应用程序
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8200)

！！！！好了 先这样吧！！！效果如何，你们尽管自己来一次看看咯！

文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

新开QQ群号，欢迎随时加群交流，相互学习。QQ群号：247491107

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822