Python & FastAPI•Part 5: ORMs, Databases & Background Workers (56 - 65)
Async SQLAlchemy with `asyncpg`:
In Simple Words:
Use create_async_engine with postgresql+asyncpg:// and AsyncSession to perform non-blocking database queries in FastAPI.
python
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session = async_sessionmaker(engine, expire_on_commit=False)
async def get_user(user_id: int):
async with async_session() as session:
result = await session.execute(select(User).where(User.id == user_id))
return result.scalar_one_or_none()