异步迭代器 ＆ 异步生成器 ＆ 异步上下文管理器

异步迭代器

类似迭代器, 实现了 __aiter__() 和 __anext__() 方法, 并且返回一个awaitable对象

obj.__aiter__(self):
    return 异步迭代器

obj.__anext__(self):
    return 迭代器下一个结果值, 结束时抛出 `StopAsyncIteration` 错误

# 示例
import asyncio

class Reader:
    def __init__(self):
        self.i = 0

    def __aiter__(self):
        return self

    async def get_next(self):
        self.i += 1
        if self.i >= 100:
            return None
        else:
            return self.i

    async def __anext__(self):
        next = await self.get_next()
        if next:
            return next
        else:
            raise StopAsyncIteration


async def main():
    # async for 需要写在协程函数中
    async for item in Reader():
        print(item)

asyncio.run(main())

异步上下文管理器

实现 __aenter__() 和 __aexit__() 的方法就是一个异步上下文管理器

import asyncio

class AsyncContextManager:
    def __init__(self):
        self.conn = None
    
    async def __aenter__(self):
        print('进入上下文...')
        self.conn = await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print('离开上下文...')
        # self.conn.close()

    async def dosomething(self):
        print('执行xxx')
        return '666'


async def main():
    async with AsyncContextManager() as f:
        res = await f.dosomething()
        print(res)


asyncio.run(main())

异步生成器

import asyncio

async def gener():
    for i in range(1, 100):
        yield i
        await asyncio.sleep(1)

async def test():
    async for item in gener():
        print(item)

async def main():
    tasks = [asyncio.create_task(test()) for _ in range(3)]
    await asyncio.wait(tasks)

asyncio.run(main())