Python Academy · レッスン

非同期コンテキストマネージャーとイテレーター

Q: 「非同期コンテキストマネージャーとイテレーター」で何を学びますか？

async with と async for のプロトコルを実装します。 ブラウザで直接実行するハンズオンコードでPython Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

async with と async for のプロトコルを実装します。

「非同期コンテキストマネージャーとイテレーター」はCoddyKit上の無料Python Academyレッスンです。これはレッスン3/4です。下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。これはPython Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 Python Academyコースには全4レッスンが含まれています。

async with文

セットアップや後処理でI/Oのawaitが必要なコンテキストマネージャーには、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__がawaitされ、例外が発生した場合でも退場時には__aexit__がawaitされます。

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())

非同期ファイルI/Oのためのaiofiles

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"))

複数の非同期コンテキストマネージャー

1行に複数の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操作を提供します。

よくある質問