您实际上在那里有一个“互锁上下文” - 如果不返回函数的__exit__部分，无论如何second它都不会恢复上下文。ContextVars
所以，我在这里想出了一些东西 - 我能想到的最好的事情是一个装饰器来显式声明哪些可调用对象将有自己的上下文 - 我创建了一个ContextLocal作为命名空间工作的类，就像thread.local- 并且该命名空间中的属性应该按照您的预期正确行事。
我现在正在完成代码 - 所以我还没有测试它的async多线程，但它应该可以工作。如果你能帮我写一个合适的测试，下面的解决方案本身就可以成为一个 Python 包。
（我不得不求助于在生成器和协同程序框架中注入一个对象，以便在生成器或协同程序结束后清理上下文注册表 - PEP 558正式化了locals()Python 3.8+ 的行为，并且我现在不记得是否允许这种注入 - 不过，它可以工作到 3.8 beta 3，所以我认为这种用法是有效的）。
无论如何，这是代码（命名为context_wrapper.py）：
"""
Super context wrapper -
meant to be simpler to use and work in more scenarios than
Python's contextvars.
Usage:
Create one or more project-wide instances of "ContextLocal"
Decorate your functions, co-routines, worker-methods and generators
that should hold their own states with that instance's `context` method -
and use the instance as namespace for private variables that will be local
and non-local until entering another callable decorated
with `intance.context` - that will create a new, separated scope
visible inside the decorated callable.
"""
import sys
from functools import wraps
__author__ = "João S. O. Bueno"
__license__ = "LGPL v. 3.0+"
class ContextError(AttributeError):
pass
class ContextSentinel:
def __init__(self, registry, key):
self.registry = registry
self.key = key
def __del__(self):
del self.registry[self.key]
_sentinel = object()
class ContextLocal:
def __init__(self):
super().__setattr__("_registry", {})
def _introspect_registry(self, name=None):
f = sys._getframe(2)
while f:
h = hash(f)
if h in self._registry and (name is None or name in self._registry[h]):
return self._registry[h]
f = f.f_back
if name:
raise ContextError(f"{name !r} not defined in any previous context")
raise ContextError("No previous context set")
def __getattr__(self, name):
namespace = self._introspect_registry(name)
return namespace[name]
def __setattr__(self, name, value):
namespace = self._introspect_registry()
namespace[name] = value
def __delattr__(self, name):
namespace = self._introspect_registry(name)
del namespace[name]
def context(self, callable_):
@wraps(callable_)
def wrapper(*args, **kw):
f = sys._getframe()
self._registry[hash(f)] = {}
result = _sentinel
try:
result = callable_(*args, **kw)
finally:
del self._registry[hash(f)]
# Setup context for generator or coroutine if one was returned:
if result is not _sentinel:
frame = getattr(result, "gi_frame", getattr(result, "cr_frame", None))
if frame:
self._registry[hash(frame)] = {}
frame.f_locals["$context_sentinel"] = ContextSentinel(self._registry, hash(frame))
return result
return wrapper
这是与它一起使用的示例的修改版本：
from contextlib import contextmanager
from context_wrapper import ContextLocal
ctx = ContextLocal()
@contextmanager
def use_mode(mode):
ctx.MODE = mode
print("entering use_mode")
print_mode()
try:
yield
finally:
pass
def print_mode():
print(f'Mode {ctx.MODE}')
@ctx.context
def first():
ctx.MODE = 0
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')
print_mode()
print("at end")
print_mode()
@ctx.context
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()
这是运行的输出：
Start first
Mode 0
entering use_mode
Mode 1
In first: with use_mode(1)
In first: start second
Start second
Mode 1
entering use_mode
Mode 2
In second: with use_mode(2)
In second: yield
In first: back from second
Mode 1
In first: continue second
In second: continue
Mode 2
In second: finish
In first: finish
Mode 1
at end
Mode 1
（它会比原生上下文变量慢几个数量级，因为它们是内置的 Python 运行时原生代码 - 但它似乎更容易包装以使用相同的数量）