标签: python-contextvars

Python 的线程本地数据和ContextVars 似乎实现了相同的目标（虽然 API 略有不同），唯一面向用户的区别是 sContextVar可以使用异步代码（协程和 asyncio），而线程本地数据则不能。

这真的是唯一的实际区别吗？

这是否意味着任何针对运行时 >= 3.7（ContextVar引入时）的代码最好使用过去ContextVar可能使用过的所有线程本地数据？或者是否有理由仍然偏爱线程本地数据？（除了您特别想要将状态与线程而不是上下文关联的情况。）

这是Python中的内存泄漏吗？

import contextvars

contextvar = contextvars.ContextVar('example')

while True:
   string = 'hello world'
   token = contextvar.set(string)

contextvar 是一个随着你向它推送的次数越多而不断增长的堆栈吗？

如果我从不打电话怎么办contextvar.reset(token)？

或者一切都通过引用计数来处理？

我对 asyncio 完全陌生ContextVars，我刚刚阅读了 3.7 中的新内容并发现ContextVars，我很难理解它的用法，我只知道它在协程中很有帮助，而不是使用thread.localshould use ContextVars。但是官方文档和顶级谷歌搜索结果都无法帮助我真正理解其目的。

那么凸变量是跨模块共享的吗？我试过：

例子.py

from contextvars import ContextVar

number = ContextVar('number', default=100)
number.set(1)

然后我尝试导入 number.py

(playground) Jamess-MacBook-Pro-2:playground jlin$ python3.7
Python 3.7.0 (v3.7.0:1bf9cc5093, Jun 26 2018, 23:26:24) 
[Clang 6.0 (clang-600.0.57)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import example
>>> from contextvars import ContextVar
>>> number = ContextVar('number', default=200)
>>> number.get()
200

我原以为number.get()会返回 1，但显然我理解它的目的是错误的。

有人可以帮我理解这一点吗？

关于以下SO答案。为了理解使用 Contextvars 和不使用 Contextvars 之间的区别，我做了一些更改。

我希望在某个时候变量myid会被破坏，但将范围更改为更高的数字似乎根本没有影响。

import asyncio
import contextvars

# declare context var
request_id = contextvars.ContextVar('Id of request.')


async def some_inner_coroutine(myid):
    # get value
    print('Processed inner coroutine of myid   : {}'.format(myid))
    print('Processed inner coroutine of request: {}'.format(request_id.get()))
    if myid != request_id.get():
        print("ERROR")


async def some_outer_coroutine(req_id):
    # set value
    request_id.set(req_id)

    await some_inner_coroutine(req_id)

    # get value
    print('Processed outer coroutine of request: {}'.format(request_id.get()))


async def main():
    tasks = []
    for req_id in range(1, 1250):
        tasks.append(asyncio.create_task(some_outer_coroutine(req_id)))

    await asyncio.gather(*tasks)


if …

将字典设置为ContextVar默认：

var: ContextVar[dict] = ContextVar('var', default={})

...有点有效，因为字典默认可用，但它始终引用相同的实例，而不是为每个上下文生成一个新实例。

contextvars 是否以某种方式支持工厂（对于字典、列表等），如下所示：

var: ContextVar[dict] = ContextVar('var', default=list)
var: ContextVar[dict] = ContextVar('var', default=lambda: dict())

或者我只需要手动完成：

var: ContextVar[Optional[dict]] = ContextVar('var', default=None)

...

if not var.get():
    var.set({})

编辑：正如指出的蒂埃里Lathuille，PEP567，在那里ContextVar被引入，却没有设计地址发生器（不像撤回PEP550）。尽管如此，主要问题仍然存在。如何编写可在多个线程，生成器和asyncio任务中正常运行的有状态上下文管理器？

我有一个带有一些可以在不同“模式”下运行的功能的库，因此可以通过本地上下文来更改其行为。我正在查看contextvars可靠地实现此功能的模块，因此可以在不同的线程，异步上下文等环境中使用它。但是，我很难获得一个正确的简单示例。考虑以下最小设置：

from contextlib import contextmanager
from contextvars import ContextVar

MODE = ContextVar('mode', default=0)

@contextmanager
def use_mode(mode):
    t = MODE.set(mode)
    try:
        yield
    finally:
        MODE.reset(t)

def print_mode():
   print(f'Mode {MODE.get()}')

这是一个带有生成器功能的小测试：

def first():
    print('Start first')
    print_mode()
    with use_mode(1):
        print('In first: with use_mode(1)')
        print('In first: start second')
        it = second()
        next(it)
        print('In first: back from second')
        print_mode()
        print('In first: continue second')
        next(it, None)
        print('In first: finish')

def second():
    print('Start second')
    print_mode() …

我正在尝试使用新的 contextvars 库（https://docs.python.org/3/library/contextvars.html）来使某些值在异步上下文中跨模块可用，类似于跨模块的 ContextVars，并且想知道是否有一种方法可以通过名称或标识符从给定的执行上下文中获取上下文变量，而无需直接导入上下文变量。

理想情况下，我想做类似以下的事情：

from _contextvars import ContextVar

language = ContextVar('language')
language.set('en-US')

在一个单独的模块中，该模块可以服务网络请求或根据设置的语言获取翻译：

from _contextvars import copy_context

ctx = copy_context()

if 'language' in ctx:
  lang_context_var = ctx.get('language') # get the context variable
  lang_value = lang_context_var.get() # get the value

我意识到 contextvars 库的 API 在 ContextVar 对象和值之间有一个映射，而不是字符串到值之间的映射，因此无法完成此查找，但我发现需要做的有点from module_with_context_var import language混乱并且容易分散在代码库中而不是当前上下文的单个包装器，这看起来像 copy_context() 应该是的。

任何能够在中央位置构造上下文变量获取和设置的好方法的指针也适用于协同例程/异步编程，我们将不胜感激！

我正在尝试在我的数据库框架中管理事务（我使用 MongoDB 和 umongo 而不是 pymongo）。

要使用事务，必须session沿整个调用链传递一个 kwarg。我想提供一个上下文管理器来隔离事务。只有调用链末尾的函数需要知道该session对象。

我发现了上下文变量，并且我已经接近一些东西，但还没有完全实现。

我想要什么：

with Transaction():\n    #\xc2\xa0Do stuff\n    d = MyDocument.find_one()\n    d.attr = 12\n    d.commit()\n

这是我现在想到的：

s = ContextVar(\'session\', default=None)\n\nclass Transaction(AbstractContextManager):\n\n    def __init__(self):\n        self.ctx = copy_context()\n        # Create a new DB session\n        session = db.create_session()\n        # Set session in context\n        self.ctx.run(s.set, session)\n\n    def __exit__(self, *args, **kwargs):\n        pass\n\n    # Adding a run method for convenience\n    def run(self, func, *args, **kwargs):\n        self.ctx.run(func, *args, **kwargs)\n\ndef func():\n    d …

在有关 Context Vars 的 Python 文档中，描述了 Context::run 方法以启用在上下文中执行可调用对象的更改，以便可调用对象对上下文执行的更改包含在复制的上下文中。如果你需要执行协程怎么办？为了实现相同的行为，你应该怎么做？

就我而言，我想要的是这样的东西来处理可能嵌套事务的事务上下文：

my_ctxvar = ContextVar("my_ctxvar")

async def coro(func, transaction):
    token = my_ctxvar.set(transaction)
    r = await func()
    my_ctxvar.reset(token)  # no real need for this, but why not either
    return r

async def foo():
    ctx = copy_context()
    # simplification to one case here: let's use the current transaction if there is one
    if tx_owner := my_ctxvar not in ctx:
        tx = await create_transaction()
    else:
        tx = my_ctxvar.get()
    
    try:
        r = await ctx.run(coro)  # not …

我目前有两个无限异步任务正在运行，并且希望在它们之间共享状态。一项任务是读取消息然后发送消息的 WebSocket 连接，另一项任务读取传入的光数据。我想在两个任务之间发送一个布尔值，表明 websocket 连接是否成功。

这是我初始化上下文的方式。

client_connect_var = contextvars.ContextVar('client_connect',default = False)
client_connect_var.set(False)
ctx = contextvars.copy_context()
async def main():
  message = json.dumps({'payload': {
                        'payload'})
  loop = asyncio.get_event_loop()
  start_light = asyncio.create_task(calculate_idle(3))
  await asyncio.gather(init_connection(message), start_light)

ctx.run(asyncio.run(main()))

这是我的 init_connection 中的代码：

async def init_connection(message):
  async with websockets.connect(uri) as websocket:
      #This should set the global context variable to true
      client_connect_var.set(True)
      CLIENT_WS = websocket
      client_connect = client_connect_var.get()
    # send init message
    await websocket.send(message)
    print("Connection is open")
    while client_connect:
        await handleMessages(websocket, message)
    await websocket.close()

这是它试图获取灯光代码中当前状态的地方

async def calculate_idle(t): …

就目前而言，我已经找到了很多关于 contextvars 模块如何与 asyncio 一起运行的示例，但没有一个关于如何与线程一起运行的示例（asyncio.get_event_loop().run_in_executor、threading.Thread 等）。

我的问题是，如何将上下文传递给单独的线程？下面您可以看到一个不起作用的代码片段（python 3.9.8）。

import typing
import asyncio
import contextvars
import concurrent.futures


class CustomThreadPoolExecutor(concurrent.futures.ThreadPoolExecutor):
    def submit(
        self,
        function: typing.Callable,
        *args,
        **kwargs
    ) -> concurrent.futures.Future:
        context = contextvars.copy_context()
        return super().submit(
            context.run,
            functools.partial(function, *args, **kwargs)
        )


def function():
    print(var.get())


async def main():
    await asyncio.get_event_loop().run_in_executor(None, function)


if __name__ == '__main__':
    var = contextvars.ContextVar('variable')
    var.set('Message.')

    asyncio.get_event_loop().set_default_executor(CustomThreadPoolExecutor)
    asyncio.run(main())