Python Academy · 课时

异步上下文管理器与迭代器

Q: 「异步上下文管理器与迭代器」这节课中我会学到什么？

实现 async with 和 async for 协议。 你通过在浏览器中直接运行的动手代码来练习 Python Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

实现 async with 和 async for 协议。

异步上下文管理器与迭代器 是 CoddyKit 上的免费 Python Academy 课时。这是第 3 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Python Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Python Academy 课程共包含 4 节课。

async with 语句

对于设置或清理过程涉及等待 I/O 的上下文管理器，请使用 async with。受管理的对象必须实现 __aenter__ 和 __aexit__。

import asyncio

class AsyncFile:
    async def __aenter__(self):
        print("open")
        return self
    async def __aexit__(self, *args):
        print("close")

async def main():
    async with AsyncFile() as f:
        print("using")

asyncio.run(main())

aenter 和 aexit

两者都是协程。进入时会等待 __aenter__；退出时会等待 __aexit__，即使发生了异常也是如此。

import asyncio

class DBSession:
    async def __aenter__(self):
        self.conn = await async_connect()
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            await self.conn.rollback()
        else:
            await self.conn.commit()
        await self.conn.close()
        return False

@asynccontextmanager

contextlib.asynccontextmanager 让您可以将异步上下文管理器编写为异步生成器函数。

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def managed_resource():
    print("acquire")
    try:
        yield {"status": "ready"}
    finally:
        print("release")

async def main():
    async with managed_resource() as r:
        print(r["status"])

asyncio.run(main())

async for 语句

async for 会遍历异步可迭代对象，也就是实现了 __aiter__ 和 __anext__ 的对象。每次迭代都可能涉及 I/O。

import asyncio

class AsyncCounter:
    def __init__(self, n): self.n, self.i = n, 0
    def __aiter__(self): return self
    async def __anext__(self):
        if self.i >= self.n: raise StopAsyncIteration
        await asyncio.sleep(0)
        self.i += 1
        return self.i

async def main():
    async for val in AsyncCounter(3):
        print(val)

asyncio.run(main())

异步生成器

包含 yield 的 async def 函数是异步生成器。请使用 async for 消费它。

import asyncio

async def ticker(n):
    for i in range(n):
        await asyncio.sleep(1)
        yield i

async def main():
    async for val in ticker(3):
        print(val)   # 0, 1, 2 with 1-s pauses

asyncio.run(main())

使用 aiofiles 进行异步文件 I/O

aiofiles 库会将文件操作封装在异步上下文管理器和异步迭代器中。

# pip install aiofiles
import aiofiles, asyncio

async def main():
    async with aiofiles.open("data.txt", "r") as f:
        async for line in f:
            print(line.strip())

aiohttp 客户端会话

aiohttp.ClientSession 是用于发出并发 HTTP 请求的异步上下文管理器。

# pip install aiohttp
import aiohttp, asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()

asyncio.run(fetch("https://api.example.com/data"))

多个异步上下文管理器

可以在一行中叠加多个 async with 子句（Python 3.10+ 的括号形式）。

import asyncio

async def main():
    async with (
        open_db() as db,
        open_cache() as cache,
    ):
        data = await db.fetch("SELECT 1")
        await cache.set("key", data)

AsyncExitStack

contextlib.AsyncExitStack 管理一组动态的异步上下文管理器。

from contextlib import AsyncExitStack
import asyncio

async def main():
    async with AsyncExitStack() as stack:
        conn1 = await stack.enter_async_context(connect("db1"))
        conn2 = await stack.enter_async_context(connect("db2"))
        await process(conn1, conn2)

使用 pytest-asyncio 测试异步代码

pytest-asyncio 让您可以使用 @pytest.mark.asyncio 标记编写异步测试函数。

# pip install pytest-asyncio
import pytest, asyncio

async def fetch(): return 42

@pytest.mark.asyncio
async def test_fetch():
    result = await fetch()
    assert result == 42

异步迭代器协议总结

如果对象具有 __aiter__，它就是一个异步可迭代对象。如果它还具有 __anext__，它就是一个异步迭代器。异步生成器会自动实现这两者。

import asyncio

async def gen():
    yield 1
    yield 2

async def main():
    g = gen()
    print(await g.__anext__())   # 1
    print(await g.__anext__())   # 2

快速检查

对象必须实现哪些双下划线方法，才能与 async with 一起使用？

回顾

对于需要异步设置或清理的资源，请使用 async with。对于异步产生值的可迭代对象，请使用 async for。异步生成器可以简化异步迭代器的创建。aiofiles 和 aiohttp 等库提供原生异步的文件和 HTTP 操作。

常见问题解答