1 更简单的全局代理

所谓的全局代理变量就是类似Flask导入request之后导出可以进行使用的方式。

之前在介绍-应用内追踪链路ID里面有讲到关于全局代理的实现，但是里面实现全局代理的时候，使用的方式比较复杂，而且是直接基于Flask的代码的方式来修改。

连接地址：fastapi框架系列-应用内追踪链路ID和全局代理request:Request，避免显示传参

后来发现另一种更简单的方式：这里大家可以参考参考哦！

实现

代码来自于index.py框架的实现。

https://github.com/index-py/index.py

1：定义一个绑定代理方法：

def bind_contextvar(contextvar):
    class ContextVarBind:
        __slots__ = ()

        def __getattr__(self, name):
            return getattr(contextvar.get(), name)

        def __setattr__(self, name, value):
            setattr(contextvar.get(), name, value)

        def __delattr__(self, name):
            delattr(contextvar.get(), name)

        def __getitem__(self, index):
            return contextvar.get()[index]

        def __setitem__(self, index, value):
            contextvar.get()[index] = value

        def __delitem__(self, index):
            del contextvar.get()[index]

    return ContextVarBind()

2：基于contextvars方式实现上下文

contextvars的是线程安全和协程安全的，使用这个最好的方式，我们可以把下面的代码定义在某个模块下。

from contextvars import ContextVar
from fastapi import Request
def bind_contextvar(contextvar):
    class ContextVarBind:
        __slots__ = ()

        def __getattr__(self, name):
            return getattr(contextvar.get(), name)

        def __setattr__(self, name, value):
            setattr(contextvar.get(), name, value)

        def __delattr__(self, name):
            delattr(contextvar.get(), name)

        def __getitem__(self, index):
            return contextvar.get()[index]

        def __setitem__(self, index, value):
            contextvar.get()[index] = value

        def __delitem__(self, index):
            del contextvar.get()[index]

    return ContextVarBind()

request_var: ContextVar[Request] = ContextVar("request")
request:Request = bind_contextvar(request_var)

3：自定义一个中间件用于初始化request_var

中间件具体代码如下：

from utils.request import request_var
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    token = request_var.set(request)
    try:
        response = await call_next(request)
        return response
    finally:
        request_var.reset(token)

4.开始进行各种导入测试验证

注意需要的是导入的是时候导入bind_contextvar(request_var)下的代理对象。

request:Request = bind_contextvar(request_var)

示例代码：

from utils.request import request
@router_org.post('/add/org', summary="机构添加")
async def callfun(*, org: Org, db_session: AsyncSession = Depends(get_db_session)):
    # 这里应该使用事务处理
    print(request.headers)
  
    return JSONResponse({
        "code": 200,
        "msg": "创建机构成功",
        "data": {"org_name": org.org_name}
    })

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

END

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

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

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

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