14 Async Database

title: Async Database Operations in FastAPI REST APIs weight: 24 date: 2026-06-28 lastmod: 2026-06-28 description: Learn async database operations in FastAPI using SQLAlchemy async support, asyncpg driver, and async ORM sessions for non-blocking database queries. tags: [api-development, fastapi]

Async database operations in FastAPI use SQLAlchemy's async engine with asyncpg for PostgreSQL, enabling non-blocking database queries that maximize throughput by not blocking the event loop during I/O.

```python
# app/database.py (async version)
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# app/routes/users.py (async)
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app import models, schemas

router = APIRouter()

@router.get("/users/{user_id}", response_model=schemas.UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(models.User).where(models.User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

What's Next

Now learn about JWT/OAuth2 authentication in Building REST APIs with FastAPI.