Python の非同期コンテキストマネージャ(__aenter__, __aexit__)

非同期コンテキストマネージャは __aenter____aexit__ を実装し、async with で使用します。非同期リソースの管理に使います。

基本構造

class AsyncResource:
    async def __aenter__(self):
        print("Acquiring resource")
        await asyncio.sleep(0.1)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Releasing resource")
        await asyncio.sleep(0.1)
        return False  # 例外を再送出

async def main():
    async with AsyncResource() as resource:
        print("Using resource")

asyncio.run(main())

出力は「Acquiring resource → Using resource → Releasing resource」の順になります。

実用例:非同期データベース接続

import asyncio

class AsyncDBConnection:
    def __init__(self, dsn):
        self.dsn = dsn
        self.connection = None
    
    async def __aenter__(self):
        print(f"Connecting to {self.dsn}")
        await asyncio.sleep(0.1)  # 接続処理
        self.connection = "connected"
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing connection")
        await asyncio.sleep(0.1)  # 切断処理
        self.connection = None
        return False
    
    async def query(self, sql):
        await asyncio.sleep(0.1)
        return f"Result of: {sql}"

async def main():
    async with AsyncDBConnection("postgres://...") as db:
        result = await db.query("SELECT * FROM users")
        print(result)

asyncio.run(main())

contextlib.asynccontextmanager

デコレータを使えば、クラスを定義せずに非同期コンテキストマネージャを作れます。

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def async_timer():
    import time
    start = time.perf_counter()
    try:
        yield
    finally:
        end = time.perf_counter()
        print(f"Elapsed: {end - start:.4f}s")

async def main():
    async with async_timer():
        await asyncio.sleep(1)

asyncio.run(main())

yield の前が __aenter__、後が __aexit__ に相当します。非同期コンテキストマネージャは、ファイル I/O、ネットワーク接続、ロックなど非同期リソースの確実な解放に役立ちます。