Async SQLAlchemy 2.0 with FastAPI — Full Setup & Async CRUD

🎯 What You Will Build

A fully async FastAPI + SQLAlchemy 2.0 stack talking to Postgres — no blocking calls, so your app can serve many concurrent requests without one slow query freezing the rest.

An async engine with asyncpg
async_sessionmaker and a reusable AsyncSession dependency
SQLAlchemy 2.0 typed models with Mapped[...] / mapped_column
Async CRUD with select() — create, list, get, update, delete
Table creation on startup with lifespan
The async gotchas that bite everyone (and how to dodge them)

Prerequisites: Python 3.11+, a running Postgres, and comfort with Depends() (see fastapi-dependency-injection). If you've only done sync SQLModel before, this is your upgrade path to true async I/O.

fastapi-async-db/
├── .env
├── requirements.txt
└── app/
    ├── __init__.py
    ├── main.py       # app, lifespan, routes
    ├── database.py   # async engine + session dependency
    ├── models.py     # SQLAlchemy 2.0 typed models
    ├── schemas.py    # Pydantic request/response models
    └── crud.py       # async data-access functions

Why Async Database Access?

A normal (sync) database call blocks the worker while it waits for Postgres to answer. During that wait, that worker can't do anything else. With async, the worker hands control back to the event loop during the wait, so it can start other requests and pick this one back up when the DB responds.

💡

Ang importante dito: async shines for I/O-bound work (DB, HTTP calls) under concurrency. It does not speed up a single query — it lets one worker juggle many requests instead of sitting idle.

⚠️

All-or-nothing rule: once you go async you must stay async on the DB path. A single blocking (sync) driver call inside an async def endpoint stalls the entire event loop — worse than sync.

Sync vs Async: What Actually Changes

Concept	Sync	Async
Driver	psycopg2 / psycopg	asyncpg
URL scheme	postgresql://	postgresql+asyncpg://
Engine	create_engine()	create_async_engine()
Session	Session / sessionmaker	AsyncSession / async_sessionmaker
`execute`	session.execute(...)	await session.execute(...)
Endpoint	def	async def
Commit	session.commit()	await session.commit()

The mental shift is small: it's the same SQLAlchemy 2.0 select() API, but you await the I/O calls and use the async engine/session variants.

Setup + .env

requirements.txt

fastapi==0.116.1
uvicorn[standard]==0.30.6
sqlalchemy==2.0.36
asyncpg==0.30.0
pydantic-settings==2.6.1

.env

# note the +asyncpg driver — this is what makes it async
DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/asyncdb

run.sh

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
# make sure the "asyncdb" database exists in Postgres first
uvicorn app.main:app --reload

💡

The +asyncpg suffix in the URL is the whole trick — it tells SQLAlchemy to use the async driver. Get this wrong and you'll see InvalidRequestError: The asyncio extension requires an async driver.

The Async Engine + Session Dependency

This is the heart of the setup. We create one async engine, one async_sessionmaker, and a yield dependency that hands an AsyncSession to any route that needs it.

app/database.py

from collections.abc import AsyncGenerator

from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase


class Settings(BaseSettings):
    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/asyncdb"
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")


settings = Settings()

# One engine for the whole app. echo=True prints SQL (turn off in prod).
engine = create_async_engine(settings.DATABASE_URL, echo=True)

# expire_on_commit=False keeps objects usable after commit (important for FastAPI).
AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


class Base(DeclarativeBase):
    """Base class for all ORM models."""


async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

⚠️

Set expire_on_commit=False. By default SQLAlchemy expires objects after commit, so reading user.id in your response would trigger a fresh (lazy, blocking) DB load and raise MissingGreenlet. Turning it off keeps the committed values in memory.

SQLAlchemy 2.0 Typed Models

SQLAlchemy 2.0 uses fully typed models with Mapped[...] and mapped_column(). This gives you editor autocomplete and mypy checks on your columns.

app/models.py

from datetime import datetime

from sqlalchemy import func
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True, index=True)
    email: Mapped[str] = mapped_column(unique=True)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

app/schemas.py

from datetime import datetime

from pydantic import BaseModel, ConfigDict, EmailStr


class UserCreate(BaseModel):
    username: str
    email: EmailStr


class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # read straight from ORM objects

    id: int
    username: str
    email: EmailStr
    is_active: bool
    created_at: datetime

💡

from_attributes=True (Pydantic v2) lets FastAPI serialize an ORM User object directly into a UserRead response — no manual dict conversion.

Create Tables on Startup (lifespan)

For a demo we create tables at startup with lifespan. Notice how even schema creation is awaited via run_sync on an async connection.

app/main.py

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import Base, engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup: create tables
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # shutdown: close the engine's connection pool
    await engine.dispose()


app = FastAPI(title="FastAPI Async DB", version="0.1.0", lifespan=lifespan)

⚠️

Demo only: create_all is fine for learning. For real projects use Alembic migrations (see fastapi-alembic-migrations) so schema changes are versioned and reviewable.

Async CRUD with select()

Keep data access in one module. Every DB call is await-ed, and reads go through select() + .scalars().

app/crud.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import User
from app.schemas import UserCreate


async def create_user(session: AsyncSession, data: UserCreate) -> User:
    user = User(username=data.username, email=data.email)
    session.add(user)
    await session.commit()
    await session.refresh(user)   # load DB-generated id/created_at
    return user


async def list_users(session: AsyncSession, limit: int = 20, offset: int = 0) -> list[User]:
    result = await session.execute(select(User).limit(limit).offset(offset))
    return list(result.scalars().all())


async def get_user(session: AsyncSession, user_id: int) -> User | None:
    return await session.get(User, user_id)


async def deactivate_user(session: AsyncSession, user: User) -> User:
    user.is_active = False
    await session.commit()
    await session.refresh(user)
    return user


async def delete_user(session: AsyncSession, user: User) -> None:
    await session.delete(user)
    await session.commit()

Now wire the routes. The AsyncSession arrives via the dependency — routes stay thin:

app/main.py

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app import crud
from app.database import get_session
from app.schemas import UserCreate, UserRead

SessionDep = Annotated[AsyncSession, Depends(get_session)]


@app.post("/users", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(data: UserCreate, session: SessionDep):
    return await crud.create_user(session, data)


@app.get("/users", response_model=list[UserRead])
async def list_users(session: SessionDep, limit: int = 20, offset: int = 0):
    return await crud.list_users(session, limit=limit, offset=offset)


@app.get("/users/{user_id}", response_model=UserRead)
async def get_user(user_id: int, session: SessionDep):
    user = await crud.get_user(session, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail=f"User {user_id} not found")
    return user


@app.patch("/users/{user_id}/deactivate", response_model=UserRead)
async def deactivate_user(user_id: int, session: SessionDep):
    user = await crud.get_user(session, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail=f"User {user_id} not found")
    return await crud.deactivate_user(session, user)


@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int, session: SessionDep):
    user = await crud.get_user(session, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail=f"User {user_id} not found")
    await crud.delete_user(session, user)

✅

Clean layering: routes handle HTTP, crud.py handles data, the dependency handles the session lifecycle. Each piece is testable on its own.

Try It (curl)

1Run and open Swagger

Start the server and open http://127.0.0.1:8000/docs. Tables are created automatically on startup.

2Exercise the full CRUD cycle

# create
$ curl -s -X POST http://127.0.0.1:8000/users \
    -H "Content-Type: application/json" \
    -d '{"username":"thirdy","email":"[email protected]"}'
{"id":1,"username":"thirdy","email":"[email protected]","is_active":true,"created_at":"2026-07-18T09:00:00"}

# list
$ curl -s http://127.0.0.1:8000/users
[{"id":1,"username":"thirdy","email":"[email protected]","is_active":true,"created_at":"..."}]

# get one
$ curl -s http://127.0.0.1:8000/users/1
{"id":1,"username":"thirdy",...}

# deactivate
$ curl -s -X PATCH http://127.0.0.1:8000/users/1/deactivate
{"id":1,"username":"thirdy","is_active":false,...}

# delete
$ curl -s -o /dev/null -w "%{http_code}\n" -X DELETE http://127.0.0.1:8000/users/1
204

✅

You did it: a non-blocking Postgres API. Under load, one Uvicorn worker can now handle many concurrent requests instead of blocking on each query.

Common Mistakes (and Fixes)

Mistake	Symptom	Fix
Using `postgresql://` (no +asyncpg)	InvalidRequestError: requires an async driver	Use `postgresql+asyncpg://`
Forgetting `await` on a DB call	Got a coroutine, not a result	Await every `execute/commit/refresh/get`
Leaving `expire_on_commit` default	MissingGreenlet when reading fields after commit	Set `expire_on_commit=False`
Accessing a lazy relationship in a response	MissingGreenlet / implicit IO error	Eager-load with `selectinload()`
Mixing a sync driver call in `async def`	Whole event loop freezes under load	Keep the entire DB path async
Creating a new engine per request	Connection pool exhaustion	One global engine; inject sessions via the dependency

💡

Relationships: lazy loading doesn't work transparently in async. When you need related rows, load them explicitly: select(User).options(selectinload(User.orders)).

What's Next

✅

🚀 Recommended next reads:

fastapi-alembic-migrations — replace create_all with versioned migrations
fastapi-database-transactions — sessions, transactions & connection pooling in depth
fastapi-error-handling — turn IntegrityError (duplicate email) into clean 409s
fastapi-redis-caching — cache hot reads so the DB does even less work

Recap: use the +asyncpg URL, one create_async_engine, an async_sessionmaker with expire_on_commit=False, and inject an AsyncSession via a yield dependency. Then await every DB call — and your FastAPI app scales with real non-blocking I/O.

Async SQLAlchemy 2.0 with FastAPI (Full Setup + Async CRUD)

🎯 What You Will Build

Why Async Database Access?

Sync vs Async: What Actually Changes

Setup + .env

The Async Engine + Session Dependency

SQLAlchemy 2.0 Typed Models

Create Tables on Startup (lifespan)

Async CRUD with select()

Try It (curl)

Common Mistakes (and Fixes)

What's Next

About the Author

Thirdy Gayares

🎯 What You Will Build

1Why Async Database Access?

2Sync vs Async: What Actually Changes

3Setup + .env

4The Async Engine + Session Dependency

5SQLAlchemy 2.0 Typed Models

6Create Tables on Startup (lifespan)

7Async CRUD with select()

8Try It (curl)

9Common Mistakes (and Fixes)

10What's Next

About the Author

Thirdy Gayares

Why Async Database Access?

Sync vs Async: What Actually Changes

Setup + .env

The Async Engine + Session Dependency

SQLAlchemy 2.0 Typed Models

Create Tables on Startup (lifespan)

Async CRUD with select()

Try It (curl)

Common Mistakes (and Fixes)

What's Next