jde*_*esa 5 python python-contextvars asyncio
编辑：正如指出的蒂埃里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()
with use_mode(2):
print('In second: with use_mode(2)')
print('In second: yield')
yield
print('In second: continue')
print_mode()
print('In second: finish')
first()
我得到以下输出：
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()
with use_mode(2):
print('In second: with use_mode(2)')
print('In second: yield')
yield
print('In second: continue')
print_mode()
print('In second: finish')
first()
据我所知，应该Mode 1代替Mode 2，因为它是从first应印刷的上下文中打印出来的use_mode(1)。但是，似乎use_mode(2)的second一直堆叠在上，直到生成器完成为止。发电机不受支持contextvars吗？如果是这样，有什么方法可以可靠地支持状态上下文管理器？可靠地说，我的意思是无论使用以下哪种方式，它都应保持一致：
asyncio实际上，您已经在那里获得了“互锁上下文” - 如果不返回__exit__函数的部分，则无论如何都second不会恢复上下文。ContextVars
所以，我在这里想出了一些东西 - 我能想到的最好的事情\是一个装饰器来显式声明哪些可调用对象将有自己的上下文 - \ n我创建了一个ContextLocal用作命名空间的类，就像thread.local- 和其中的属性命名空间应该按照您的预期正常运行。
我现在正在完成代码 - 所以我还没有测试它的多线程async或多线程，但它应该可以工作。如果你能帮我编写一个适当的测试，下面的解决方案本身就可以成为一个Python包。
（我不得不在生成器和协同例程框架本地字典中注入一个对象，以便在生成器或协同例程结束后清理上下文注册表 - PEP 558locals()正式化了Python 3.8+的行为，并且我现在不记得是否允许这种注入 - 不过它最多可以工作到 3.8 beta 3，所以我认为这种用法是有效的）。
无论如何，这是代码（名为context_wrapper.py）：
"""\nSuper context wrapper -\n\nmeant to be simpler to use and work in more scenarios than\nPython\'s contextvars.\n\nUsage:\nCreate one or more project-wide instances of "ContextLocal"\nDecorate your functions, co-routines, worker-methods and generators\nthat should hold their own states with that instance\'s `context` method -\n\nand use the instance as namespace for private variables that will be local\nand non-local until entering another callable decorated\nwith `intance.context` - that will create a new, separated scope\nvisible inside the decorated callable.\n\n\n"""\n\nimport sys\nfrom functools import wraps\n\n__author__ = "Jo\xc3\xa3o S. O. Bueno"\n__license__ = "LGPL v. 3.0+"\n\nclass ContextError(AttributeError):\n pass\n\n\nclass ContextSentinel:\n def __init__(self, registry, key):\n self.registry = registry\n self.key = key\n\n def __del__(self):\n del self.registry[self.key]\n\n\n_sentinel = object()\n\n\nclass ContextLocal:\n\n def __init__(self):\n super().__setattr__("_registry", {})\n\n def _introspect_registry(self, name=None):\n\n f = sys._getframe(2)\n while f:\n h = hash(f)\n if h in self._registry and (name is None or name in self._registry[h]):\n return self._registry[h]\n f = f.f_back\n if name:\n raise ContextError(f"{name !r} not defined in any previous context")\n raise ContextError("No previous context set")\n\n\n def __getattr__(self, name):\n namespace = self._introspect_registry(name)\n return namespace[name]\n\n\n def __setattr__(self, name, value):\n namespace = self._introspect_registry()\n namespace[name] = value\n\n\n def __delattr__(self, name):\n namespace = self._introspect_registry(name)\n del namespace[name]\n\n def context(self, callable_):\n @wraps(callable_)\n def wrapper(*args, **kw):\n f = sys._getframe()\n self._registry[hash(f)] = {}\n result = _sentinel\n try:\n result = callable_(*args, **kw)\n finally:\n del self._registry[hash(f)]\n # Setup context for generator or coroutine if one was returned:\n if result is not _sentinel:\n frame = getattr(result, "gi_frame", getattr(result, "cr_frame", None))\n if frame:\n self._registry[hash(frame)] = {}\n frame.f_locals["$context_sentinel"] = ContextSentinel(self._registry, hash(frame))\n\n return result\n return wrapper\n这是您的示例的修改版本：
\nfrom contextlib import contextmanager\n\nfrom context_wrapper import ContextLocal\n\nctx = ContextLocal()\n\n\n@contextmanager\ndef use_mode(mode):\n ctx.MODE = mode\n print("entering use_mode")\n print_mode()\n try:\n yield\n finally:\n\n pass\n\ndef print_mode():\n print(f\'Mode {ctx.MODE}\')\n\n\n@ctx.context\ndef first():\n ctx.MODE = 0\n print(\'Start first\')\n print_mode()\n with use_mode(1):\n print(\'In first: with use_mode(1)\')\n print(\'In first: start second\')\n it = second()\n next(it)\n print(\'In first: back from second\')\n print_mode()\n print(\'In first: continue second\')\n next(it, None)\n print(\'In first: finish\')\n print_mode()\n print("at end")\n print_mode()\n\n@ctx.context\ndef second():\n print(\'Start second\')\n print_mode()\n with use_mode(2):\n print(\'In second: with use_mode(2)\')\n print(\'In second: yield\')\n yield\n print(\'In second: continue\')\n print_mode()\n print(\'In second: finish\')\n\nfirst()\n这是运行的输出：
\nStart first\nMode 0\nentering use_mode\nMode 1\nIn first: with use_mode(1)\nIn first: start second\nStart second\nMode 1\nentering use_mode\nMode 2\nIn second: with use_mode(2)\nIn second: yield\nIn first: back from second\nMode 1\nIn first: continue second\nIn second: continue\nMode 2\nIn second: finish\nIn first: finish\nMode 1\nat end\nMode 1\n（它会比本机上下文变量慢几个数量级，因为它们是内置的 Python 运行时本机代码 - 但似乎更容易以相同的数量使用）
\n