当生成器被包装为上下文管理器时,使用 Python 生成器的 .send()

Use Python generator's .send() when generator is wrapped to act as a context manager

Python 的 contextlib 提供包装器以将生成器转变为上下文管理器:

from contextlib import contextmanager

@contextmanager
def gen():
    yield

with gen() as cm:
    ...

并且生成器提供了将值发送到刚刚产生的生成器的能力:

def gen():
    x = yield
    print(x)

g = gen()
next(g)
g.send(1234) # prints 1234 and raises a StopIteration because we have only 1 yield

有没有办法同时获得这两种行为?我想将一个值发送到我的上下文管理器中,以便在处理 __exit__ 时可以使用它。所以像这样:

from contextlib import contextmanager

@contextmanager
def gen():
    x = yield
    # do something with x

with gen() as cm:
    # generator already yielded from __enter__, so we can send
    something.send(1234)

我不确定这是否是一个 good/reasonable 想法。 我觉得它确实打破了一些抽象层,因为我假设上下文管理器是作为包装生成器实现的。

如果这是一个可行的想法,我不确定 something 应该是什么。

我不认为这是可能的,可能有更好的方法来实现你想要的。

顺便说一下contextmanager is implemented is described in PEP 344with的说法。 contextmanager 在用户的生成器函数上调用 next 方法两次,一次在 __enter__ 方法中,一次在 __exit__ 方法中,两次都没有发送任何值。

您实现所需内容的方式大致如下(未经测试):

def create_context_manager(value_you_want_to_pass):
    @contextmanager
    def gen():
        do_something_with(value_you_want_to_pass)
        yield the_object_for_with_statement
        do_something_else_with(value_you_want_to_pass)
    return gen

with create_context_manager(1234) as cm:
    do_something_with(cm)  
    

@contextmanager 下的生成器可通过其 gen 属性直接访问。由于生成器无法访问上下文管理器,后者必须存储在上下文之前:

from contextlib import contextmanager

@contextmanager
def gen():
    print((yield))  # first yield to suspend and receive value...
    yield           # ... final yield to return at end of context

manager = gen()     # manager must be stored to keep it accessible
with manager as value:
    manager.gen.send(12)

生成器具有恰到好处的 yield 点数很重要 - @contextmanager 确保生成器在退出上下文后耗尽。

@contextmanager.throw 在上下文中引发异常,并且 .send None 完成后,底层生成器可以侦听:

@contextmanager
def gen():
    # stop when we receive None on __exit__
    while (value := (yield)) is not None:
        print(value)

不过,在许多情况下,将上下文管理器实现为自定义 class 可能更容易。这避免了对 send/recv 值和 pause/resume 上下文使用相同通道的复杂性。

class SendContext:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    def send(self, value):
        print("We got", value, "!!!")


with SendContext() as sc:
    sc.send("Deathstar")