如何将父线程的ContextVars传递给threading.Thread创建的子线程？

不要在每个模块单独创建ContextVar实例，而是在一个公共模块中定义所有需要的上下文变量，其他模块直接导入使用：

# context_vars.py (公共模块)
import contextvars

# 定义全局上下文变量
current_user = contextvars.ContextVar("current_user", default=None)
request_id = contextvars.ContextVar("request_id", default=None)

其他模块使用时直接导入：

# module_a.py
from context_vars import current_user

def set_user(user):
    current_user.set(user)

def get_user():
    return current_user.get()

# module_b.py
from context_vars import current_user

def print_user():
    print(f"Current user: {current_user.get()}")

这样单线程下跨模块访问完全正常，无需依赖Werkzeug的Local类，更符合contextvars的原生设计。

contextvars的上下文不会自动被子线程继承，需要手动保存父线程的上下文快照，在子线程启动时恢复：

import threading
import contextvars

class ContextThread(threading.Thread):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 保存父线程的上下文快照
        self.parent_context = contextvars.copy_context()

    def run(self):
        # 在子线程中恢复父线程的上下文
        with self.parent_context:
            super().run()

使用示例：

from context_vars import current_user
from module_b import print_user

def main():
    current_user.set("Alice")
    print("Main thread user:", current_user.get())  # 输出 Alice

    # 创建带上下文传递的子线程
    t = ContextThread(target=print_user)
    t.start()
    t.join()  # 子线程输出 Current user: Alice

基于concurrent.futures.ThreadPoolExecutor封装，在提交任务时保存当前上下文，任务执行时恢复：

from concurrent.futures import ThreadPoolExecutor
import contextvars

class ContextThreadPoolExecutor(ThreadPoolExecutor):
    def submit(self, fn, *args, **kwargs):
        # 保存提交任务时的上下文
        current_context = contextvars.copy_context()
        # 包装任务函数，先恢复上下文再执行原函数
        def wrapped_fn(*args, **kwargs):
            with current_context:
                return fn(*args, **kwargs)
        return super().submit(wrapped_fn, *args, **kwargs)

使用示例：

from context_vars import current_user
from module_b import print_user

def main():
    current_user.set("Bob")
    print("Main thread user:", current_user.get())  # 输出 Bob

    with ContextThreadPoolExecutor(max_workers=2) as executor:
        future = executor.submit(print_user)
        future.result()  # 输出 Current user: Bob

如果仍需要保留基于Werkzeug Local的实现，也可以在子线程中手动传递Local的线程数据，但contextvars是Python 3.7+的标准库，原生支持上下文管理，比Werkzeug Local更适合这种场景。

内容的提问来源于stack exchange，提问作者Abhay