Async context managers manage async database connections automatically.
1import asyncio2from contextlib import asynccontextmanager34# Using aiopg (async PostgreSQL)5import aiopg67@asynccontextmanager8async def get_connection(dsn):9 conn = await aiopg.connect(dsn)10 try:11 yield conn12 finally:13 await conn.close()1415async def main():16 async with get_connection("dbname=mydb") as conn:17 async with conn.cursor() as cur:18 await cur.execute("SELECT * FROM users")19 rows = await cur.fetchall()20 print(rows)2122asyncio.run(main())
SQLAlchemy async:
1from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine23engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")45async def get_session():6 async with AsyncSession(engine) as session:7 yield session
Benefits: