非同期コンテキストマネージャは __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、ネットワーク接続、ロックなど非同期リソースの確実な解放に役立ちます。
