Async context managers use async with for resource management in async code.
1import asyncio23class AsyncDB:4 async def __aenter__(self):5 print("Connecting...")6 await asyncio.sleep(1) # Simulate async connection7 return self89 async def __aexit__(self, exc_type, exc_val, exc_tb):10 print("Disconnecting...")11 await asyncio.sleep(0.5)12 return False1314 async def query(self, sql):15 await asyncio.sleep(0.1)16 return f"Result: {sql}"1718async def main():19 async with AsyncDB() as db:20 result = await db.query("SELECT * FROM users")21 print(result)2223asyncio.run(main())
Using contextlib:
1from contextlib import asynccontextmanager23@asynccontextmanager4async def managed_resource():5 resource = await acquire()6 try:7 yield resource8 finally:9 await release(resource)