Async Context Managers

Introduction

Async context managers implement aenter and aexit for use with async with. They're essential for managing resources in async code.

Basic Async Context Manager

class AsyncResource:
    async def __aenter__(self):
        await asyncio.sleep(0.1)  # Simulate connection
        print("Resource acquired")
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await asyncio.sleep(0.1)  # Simulate cleanup
        print("Resource released")

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

Async Context Manager with Error Handling

class DatabaseConnection:
    async def __aenter__(self):
        self.conn = await connect_to_db()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            await self.conn.rollback()
        await self.conn.close()
        return False

async def main():
    async with DatabaseConnection() as db:
        await db.execute("INSERT INTO ...")

asynccontextmanager

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_connection(url):
    conn = await connect(url)
    try:
        yield conn
    finally:
        await conn.close()

async def main():
    async with managed_connection("http://api.example.com") as client:
        response = await client.get("/data")

Multiple Async Context Managers

from contextlib import AsyncExitStack

async def main():
    async with AsyncExitStack() as stack:
        conn1 = await stack.enter_async_context(connect("db1"))
        conn2 = await stack.enter_async_context(connect("db2"))
        # Both will be properly cleaned up

Resource Pool Pattern

class Pool:
    def __init__(self, size):
        self.size = size
        self.available = []
        self.in_use = []
    
    async def __aenter__(self):
        for _ in range(self.size):
            self.available.append(await create_resource())
        return self
    
    async def __aexit__(self, *args):
        for r in self.available + self.in_use:
            await r.close()

async def main():
    async with Pool(5) as pool:
        resource = pool.available.pop()
        # Use resource

Practice Problems

1. Create async context manager for network connection
2. Handle exceptions in aexit
3. Use asynccontextmanager decorator
4. Implement async resource pool
5. Use AsyncExitStack for multiple resources