Python typing 怎样为异步生成器标注返回值

给异步生成器写类型注解时，最容易混淆的是“它会产出什么”和“它最终返回什么”。只要函数体里有 yield，它就是异步生成器，调用后得到异步迭代器，调用方应使用 async for；它不能像普通协程那样用 await 取得一个最终返回值。

先看函数体：有 yield 就不是普通协程

async def 本身不能决定返回注解。下面两个函数都以 async def 开头，但调用方式完全不同：前者逐项产出数据，后者等待完成后返回一个列表。

from collections.abc import AsyncGenerator

async def chunks() -> AsyncGenerator[bytes, None]:
    # 每次 yield 产出一块 bytes；这里不接收调用方通过 asend 送入的值。
    yield b"part-1"
    yield b"part-2"

async def all_chunks() -> list[bytes]:
    # 协程最终一次性返回列表，调用方应使用 await。
    return [b"part-1", b"part-2"]

async for chunk in chunks():
    # 异步生成器的消费入口是 async for。
    print(chunk)

items = await all_chunks()  # 协程才通过 await 得到返回值

判断标准是函数体里是否出现 yield，而不是标题里的“异步”两个字。Python 官方文档把这种函数定义为 asynchronous generator function；它结束时产生的是 StopAsyncIteration，不是一个可供调用方读取的普通返回值。

AsyncGenerator 的两个类型参数分别代表什么

在 Python 3.9 及以上，推荐从 collections.abc 导入抽象类型：

from collections.abc import AsyncGenerator

async def numbers(limit: int) -> AsyncGenerator[int, None]:
    # 第一个参数 int 表示每次 yield 的元素类型。
    # 第二个参数 None 表示不接受 agen.asend(value) 送入的业务值。
    for number in range(limit):
        yield number

async def controllable() -> AsyncGenerator[int, str]:
    # 这里演示发送类型：第一次启动仍需发送 None，后续可发送 str。
    command = yield 0
    if command == "stop":
        return
    yield 1

AsyncGenerator[T, S] 的 T 是读者从 async for 得到的类型，S 是通过异步生成器对象的 asend() 送回函数内部的类型。多数数据流只向外发数据，不需要双向通信，因此第二个参数写 None 最准确。若项目仍使用 typing.AsyncGenerator，它也能表达同一含义，但现代 Python 文档将它视为 collections.abc.AsyncGenerator 的兼容别名。

不接收发送值时，AsyncIterator 更适合作为公开接口

如果实现内部确实不会调用 asend()，公开函数可以写成 AsyncIterator[T]。它只承诺“这是一个可以异步遍历的 T 序列”，不会把生成器的双向控制能力暴露给调用方。

from collections.abc import AsyncIterable, AsyncIterator

async def parse_lines(source: AsyncIterable[str]) -> AsyncIterator[dict[str, str]]:
    # 输入也是异步可迭代对象，输出只承诺逐条产生字典。
    async for line in source:
        key, value = line.split("=", 1)
        # 具体格式错误交给调用方决定是记录、跳过还是终止。
        yield {"key": key.strip(), "value": value.strip()}

实现可以返回具体的异步生成器，注解却可以选择更窄的 AsyncIterator。这类似于把函数参数写成抽象接口：调用方只依赖真正需要的能力，后续把实现替换成自定义异步迭代器时，改动更小。

异步生成器没有“最终返回值”，汇总结果要单独传递

异步生成器中可以写空的 return，但不能写 return summary。如果既要逐项输出，又要在最后得到统计结果，可以把统计对象作为最后一项明确的数据类型，或者将统计动作拆成另一个协程：

from collections.abc import AsyncIterable, AsyncIterator
from dataclasses import dataclass

@dataclass
class Record:
    # 用联合数据对象区分普通记录与汇总记录。
    kind: str
    value: str | int

async def records() -> AsyncIterator[Record]:
    # 读者可以根据 kind 判断最后一项是否为汇总信息。
    yield Record("item", "alpha")
    yield Record("item", "beta")
    yield Record("summary", 2)
    return  # 只能无值结束，不能 return Record(...)

async def count_records(source: AsyncIterable[str]) -> int:
    # 需要最终数字时，使用独立协程直接返回 int。
    count = 0
    async for _ in source:
        count += 1
    return count

相关问题

AsyncGenerator 能不能写第三个 ReturnType 参数？

不能。它只有产出类型和发送类型两个参数，不像同步 Generator 那样额外描述返回值。

什么时候应该写 AsyncIterable 而不是 AsyncIterator？

当函数只需要被异步遍历、并不要求调用方直接使用 __anext__() 时，AsyncIterable[T] 可以继续收窄依赖。

typing.AsyncGenerator 还能不能用？

可以用于兼容旧代码，但新代码优先从 collections.abc 导入对应抽象类型，并让项目的类型检查器按目标 Python 版本配置。