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、ネットワーク接続、ロックなど非同期リソースの確実な解放に役立ちます。
49views
