如何在Python中定义“空”异步生成器？

Question

我需要定义一个“空”异步生成器，即不生成任何值的生成器。由于异步生成器的定义是包含yield语句的异步函数，这就是我想到的：

async def _empty_async_generator() -> AsyncGenerator[str, None]:
    if False:
        yield ""

这是 python 3.10 中可以做到的最好的吗？是否有用于创建空异步生成器的库函数？

Answer 1

您的代码是正确的。 （目前）没有更好的方法来做到这一点。至少不会让代码变得难以阅读。

以下是不同异步生成器的一些示例，它们可能会为问题添加一些上下文，包括一个

yield from

很方便的示例（尽管现在不可能）：

from asyncio import run
from typing import AsyncGenerator


async def empty_generator(
    do_something: bool = True,
) -> AsyncGenerator[int, None]:
    """
    A generator that is conditionally "empty".
    """
    print(f"{do_something=}")
    if not do_something:
        return
    yield 10


async def completely_empty_generator() -> AsyncGenerator[int, None]:
    """
    A generator that is always empty.

    "yield" is required to make this a generator. Without it, the function
    would be a coroutine function that returns None.
    """
    if False:
        yield 10


async def nested_generator() -> AsyncGenerator[int, None]:
    """
    A helper generator that yields numbers from 0 to 9.

    It is used to demonstrate the behavior of consuming a generator within
    another generator that is conditionally empty. This in example for the
    "yield from" portion of the quetion.
    """
    for i in range(10):
        yield i


async def consuming_nested_generator(
    disabled: bool,
) -> AsyncGenerator[int, None]:
    """
    A generator that consumes another generator conditionally.

    We have to be careful here:

    1. We can't return "None". This is not allowed here.
    2. We can't use "yield from" because it would raise a SyntaxError.
    """
    if disabled:
        return
    async for i in nested_generator():
        yield i


async def main() -> None:
    """
    The main entry point.

    This demonstrates the examples above.
    """
    async for i in empty_generator():
        print("active", i)
    async for i in empty_generator(False):
        print("inactive", i)
    async for i in completely_empty_generator():
        print("noop", i)
    async for i in consuming_nested_generator(False):
        print("nested disabled", i)
    async for i in consuming_nested_generator(True):
        print("nested enabled", i)


run(main())