да.tricky.
У вас там действительно есть "взаимосвязанный контекст" - без возврата части __exit__ для функции second он не восстановит контекст с помощью ContextVars, независимо от того, что
Итак, я придумал кое-что здесь - и лучшее, что я мог придумать, - это декоратор для явного объявления , у которого вызываемые объекты будут иметь свой собственный контекст - я создал класс ContextLocal, который работает как пространство имен, jsutкак thread.local - и атрибуты в этом пространстве имен должны работать должным образом, как вы ожидаете.
Сейчас я заканчиваю код - поэтому я еще не проверял его на асинхронность или многопоточность, но он должен работать - еслиВы можете помочь мне написать правильную тестовую батарею, решение ниже может стать пакетом Python.
(Мне пришлось прибегнуть к внедрению объекта в словарь локальных кадров фреймов генератора и сопрограмм, чтобы очистить реестр контекста после завершения работы генератора или сопрограммы - есть PEP 558 формализует поведение locals () для Python 3.8 и более поздних версий, и я не помню сейчас, разрешена ли эта инъекция - она работает до 3,8 бета 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 - но, кажется, проще обернуть-разум использовать столько же)