Python异步迭代器协议

【Python 异步迭代器协议】
==========================

异步迭代器通过 __aiter__ 和 __anext__ 协议实现，配合
async for 循环使用。还包括异步生成器和异步推导式。

──────────────────────────────────────────────
示例 1：自定义异步迭代器
──────────────────────────────────────────────

import asyncio

class 异步数字范围:
"""异步迭代器，模拟 range 但每次 yield 前异步等待"""

def __init__(self, 开始, 结束, 延迟=0.5):
self.当前 = 开始
self.结束 = 结束
self.延迟 = 延迟

def __aiter__(self):
"""返回异步迭代器自身"""
print("开始异步迭代")
return self

async def __anext__(self):
"""异步获取下一个值"""
if self.当前 >= self.结束:
raise StopAsyncIteration # 停止迭代

await asyncio.sleep(self.延迟) # 模拟异步操作
值 = self.当前
self.当前 += 1
return 值

async def 自定义迭代器示例():
"""使用 async for 遍历异步迭代器"""
async for 数字 in 异步数字范围(1, 5, 0.3):
print(f"获取到: {数字}")

asyncio.run(自定义迭代器示例())

──────────────────────────────────────────────
示例 2：异步生成器
──────────────────────────────────────────────

import asyncio

async def 异步斐波那契(上限):
"""异步生成器：生成斐波那契数列"""
a, b = 0, 1
while a < 上限:
await asyncio.sleep(0.2) # 每次生成前异步等待
yield a
a, b = b, a + b

async def 异步数据流():
"""模拟从外部源获取数据流"""
for i in range(10):
await asyncio.sleep(0.1)
yield f"数据包-{i}"

async def 生成器示例():
"""使用 async for 消费异步生成器"""
print("=== 斐波那契数列 ===")
async for 值 in 异步斐波那契(20):
print(f"斐波那契: {值}")

print("\n=== 数据流 ===")
async for 数据 in 异步数据流():
print(f"收到: {数据}")

asyncio.run(生成器示例())

──────────────────────────────────────────────
示例 3：异步列表推导式
──────────────────────────────────────────────

import asyncio

async def 异步乘法器(因子):
"""异步生成器"""
for i in range(5):
await asyncio.sleep(0.1)
yield i * 因子

async def 异步推导式示例():
"""使用异步推导式收集结果"""

# 异步列表推导式
结果 = [值 async for 值 in 异步乘法器(3)]
print(f"列表推导: {结果}")

# 异步集合推导式
结果集合 = {值 async for 值 in 异步乘法器(2)}
print(f"集合推导: {结果集合}")

# 异步字典推导式
结果字典 = {
f"键-{值}": 值
async for 值 in 异步乘法器(5)
if 值 > 5 # 支持条件过滤
}
print(f"字典推导: {结果字典}")

# 异步生成器表达式（返回异步生成器）
异步生成器 = (值 async for 值 in 异步乘法器(4))
async for 值 in 异步生成器:
print(f"生成器表达式: {值}")

asyncio.run(异步推导式示例())

──────────────────────────────────────────────
示例 4：异步迭代器实现分页
──────────────────────────────────────────────

import asyncio

class 异步分页器:
"""模拟异步分页 API 数据获取"""

def __init__(self, 总页数, 每页大小=10):
self.总页数 = 总页数
self.每页大小 = 每页大小
self.当前页 = 0

async def _获取页(self, 页码):
"""模拟异步 API 调用获取一页数据"""
await asyncio.sleep(0.3)
return [
f"数据-{页码}-{i}" for i in range(self.每页大小)
]

def __aiter__(self):
return self

async def __anext__(self):
if self.当前页 >= self.总页数:
raise StopAsyncIteration
self.当前页 += 1
数据 = await self._获取页(self.当前页)
return self.当前页, 数据

async def 分页示例():
"""使用 async for 遍历分页数据"""
async for 页码, 数据 in 异步分页器(3, 5):
print(f"第{页码}页: 共{len(数据)}条")
for 项 in 数据[:2]: # 只显示前 2 条
print(f" {项}")

asyncio.run(分页示例())

──────────────────────────────────────────────
核心要点
──────────────────────────────────────────────
• __aiter__ - 返回异步迭代器对象
• __anext__ - 返回 awaitable，结束时抛 StopAsyncIteration
• async for - 遍历异步迭代器的语法糖
• async def 生成器 + yield - 异步生成器
• async for...in...if 推导式 - 异步推导式
• 异步迭代器适合分页查询、数据流、事件流等场景