1 在新开线程里使用全局代理request遇到异常问题LookupError: <ContextVar name='

关于全局代理request的说明可以参考前面文章所诉，这里不再重复，可以查看此文章：

fastapi框架系列-再谈request body重复消费问题简析，推翻以前一些案例说明

近期有网友问起，在使用我全局代理request这种（最新方案）方式之后，如果使用在自定义新开的线程里面直接的使用的话，则会遇到问题，具体问题如下图示：

而需要包的异步函数任务如下如代码所示：


def run(n):
    # print(request.state)
    print("task",n)
    # print("funcfuncfunc",func)
    print(request.state)
    print(request)
    print(request_var.get())


    print('3s')

在上面的函数中我们的使用了自己定义全局的request。再经过自己实践之后确实会发现会有这种情况，由于我的业务需求可能和他不一样，通常我们直接把一些异步的任务放置到后台任务就可以，然后再后台任务重我依然可以正常的使用全局代理request。如下代码所示：

经过分析对比，为什么自己开的线程就会遇到问题呢？下面我们就探究一下BackgroundTasks。

BackgroundTasks内在逻辑

源码简单分析：

import asyncio
import typing

from starlette.concurrency import run_in_threadpool


class BackgroundTask:
    def __init__(
        self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
    ) -> None:
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.is_async = asyncio.iscoroutinefunction(func)

    async def __call__(self) -> None:
        if self.is_async:
            await self.func(*self.args, **self.kwargs)
        else:
            await run_in_threadpool(self.func, *self.args, **self.kwargs)


class BackgroundTasks(BackgroundTask):
    def __init__(self, tasks: typing.Sequence[BackgroundTask] = None):
        self.tasks = list(tasks) if tasks else []

    def add_task(
        self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
    ) -> None:
        task = BackgroundTask(func, *args, **kwargs)
        self.tasks.append(task)

    async def __call__(self) -> None:
        for task in self.tasks:
            await task()

BackgroundTask是具体的后台任务封装，里面主要是保存具体的函数和参数，还判断是否协程或非协程函数的处理，当执行对象的（）的时候调用__call__根据不同的函数类型执行不同的处理方式。
BackgroundTasks 是对所有BackgroundTask后台任务的集合封装

长上面分析我们可以提取到，我们值需要分析run_in_threadpool的代码即可:

async def run_in_threadpool(
    func: typing.Callable[..., T], *args: typing.Any, **kwargs: typing.Any
) -> T:
    if contextvars is not None:  # pragma: no cover
        # Ensure we run in the same context
        child = functools.partial(func, *args, **kwargs)
        context = contextvars.copy_context()
        func = context.run
        args = (child,)
    elif kwargs:  # pragma: no cover
        # run_sync doesn't accept 'kwargs', so bind them in here
        func = functools.partial(func, **kwargs)
    return await anyio.to_thread.run_sync(func, *args)

在里面他使用到了contextvars的.copy_context()复制一个context，并且把当前要执行的异步任务做了分成处理：具体处理方式如下核心代码：

child = functools.partial(func, *args, **kwargs)
context = contextvars.copy_context()
func = context.run

所以大概明白了，copy_context主要目的复制当前的上下文环境对象，然后使用当前上下文对象的中在run中运行具体callback函数，此时所有的上下文环境变量的修改也都会记录到当前复制出来context对象中。网友他按我给的提示用上面这种方式可以解决他的问题了！为了记录下这种问题，我自己也亲自实践一下。所以按部就班的方式，我们可以采取这种方式来在新开线程处理，最终我们自定义一个线程类，如下代码所示：

from threading import Thread
class MyThread(Thread):
    def __init__(self, target=None, args=(), kwargs={}):
        super().__init__()
        try:
            import contextvars  # Python 3.7+ only or via contextvars backport.
        except ImportError:  # pragma: no cover
            contextvars = None  # type: ignore
        if contextvars is not None:  # pragma: no cover
            # Ensure we run in the same context
            child = functools.partial(target, *args, **kwargs)
            context = contextvars.copy_context()
            func = context.run
            args = (child,)
        elif kwargs:  # pragma: no cover
            # run_sync doesn't accept 'kwargs', so bind them in here
            func = functools.partial(target, **kwargs)
        else:
            func = functools.partial(target, **kwargs)

        self.target = func
        self.args = args
        self.kwargs = kwargs

    def run(self):
        self.target(*self.args, **self.kwargs)

PS:上面代码其实还有待优化，主要是考虑协程和协程的问题

然后调用方式如下：

@app.get("/")
def ordercallback(*, background_tasks: BackgroundTasks):
    # print(request)

    # th = threading.Thread(target=run,name="thread_1" ,args = ("thread 1" , ))
    # th.setDaemon ( True)
    # th.start()

    th = MyThread(target=run,args=(1,))
    th.start()
    # background_tasks.add_task(run,'2345345')
    return ""

最终我们在执行异步任务run的时候，里面使用全局代理request也可以正常使用了：

def run(n):
    # print(request.state)
    print("task",n)
    # print("funcfuncfunc",func)
    print(request.state)
    print(request)
    print(request_var.get())
    print('1111')

完整示例代码如下：

from fastapi import FastAPI
from fastapi import BackgroundTasks

app = FastAPI()

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)


# 把子进程设置为守护线程，必须在start()之前设置



from concurrent.futures import ThreadPoolExecutor
import functools
thread_pool = ThreadPoolExecutor(max_workers=2)


import threading

from starlette.concurrency import run_in_threadpool
@app.get("/")
def ordercallback(*, background_tasks: BackgroundTasks):
    # print(request)

    # th = threading.Thread(target=run,name="thread_1" ,args = ("thread 1" , ))
    # th.setDaemon ( True)
    # th.start()

    th = MyThread(target=run,args=(1,))
    th.start()
    # background_tasks.add_task(run,'2345345')
    return ""


from threading import Thread
class MyThread(Thread):
    def __init__(self, target=None, args=(), kwargs={}):
        super().__init__()
        try:
            import contextvars  # Python 3.7+ only or via contextvars backport.
        except ImportError:  # pragma: no cover
            contextvars = None  # type: ignore
        if contextvars is not None:  # pragma: no cover
            # Ensure we run in the same context
            child = functools.partial(target, *args, **kwargs)
            context = contextvars.copy_context()
            func = context.run
            args = (child,)
        elif kwargs:  # pragma: no cover
            # run_sync doesn't accept 'kwargs', so bind them in here
            func = functools.partial(target, **kwargs)
        else:
            func = functools.partial(target, **kwargs)

        self.target = func
        self.args = args
        self.kwargs = kwargs

    def run(self):
        self.target(*self.args, **self.kwargs)

def run(n):
    # print(request.state)
    print("task",n)
    # print("funcfuncfunc",func)
    print(request.state)
    print(request)
    print(request_var.get())


    print('3s')


@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:
        pass
        request_var.reset(token)


if __name__ == "__main__":
    # from framecore.utils.routes_helper import print_all_routes_info
    # print_all_routes_info(app)
    import uvicorn
    import os

    app_modeel_name = os.path.basename(__file__).replace(".py", "")
    # print(app_modeel_name)
    uvicorn.run(f"{app_modeel_name}:app", host='127.0.0.1', port=31100, reload=True, workers=3)

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

结尾

END

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

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

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

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

ITPub博客

fastapi框架系列- 新开线程用全局request遇到LookupError:

1 在新开线程里使用全局代理request遇到异常问题LookupError: <ContextVar name='

BackgroundTasks内在逻辑

结尾

END

最新文章