FastAPI + SQLAlchemy + PostgreSQL: A Modern Setup

FastAPI gives you a beautiful API layer. SQLAlchemy 2.0 gives you a great ORM with first-class async support. PostgreSQL gives you a database you can actually trust. Wired together, they’re one of the most productive backend stacks available in Python today.

This post is a practical walkthrough — not a “hello world” — for setting them up properly. By the end you’ll have a project structure that scales, async DB sessions injected the right way, and Alembic migrations under version control.

What we’re building

A small “tasks” API:

POST /tasks/ — create a task
GET /tasks/ — list tasks (with pagination)
GET /tasks/{id} — get one task
PATCH /tasks/{id} — update a task
DELETE /tasks/{id} — delete a task

Backed by PostgreSQL via SQLAlchemy 2.0 in async mode, with Alembic-managed migrations.

Project setup

mkdir tasks-api && cd tasks-api
uv init
uv add fastapi[standard] sqlalchemy[asyncio] asyncpg pydantic-settings alembic

If uv isn’t your thing, plain pip install works the same way. (See Python Virtual Environments for the tooling tradeoff.)

Project structure:

tasks-api/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py
│   │   └── database.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── task.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── task.py
│   └── api/
│       ├── __init__.py
│       ├── deps.py
│       └── routes/
│           ├── __init__.py
│           └── tasks.py
├── alembic/
│   ├── env.py
│   └── versions/
├── alembic.ini
└── pyproject.toml

Configuration with `pydantic-settings`

Hard-coded config is a footgun. Use pydantic-settings to load from env vars:

# app/core/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    app_name: str = "tasks-api"
    debug: bool = False
    database_url: str  # e.g. postgresql+asyncpg://user:pass@localhost/tasks


@lru_cache
def get_settings() -> Settings:
    return Settings()

Create a .env file:

DATABASE_URL=postgresql+asyncpg://tasksuser:tasksdbpass@localhost:5432/tasksdb
DEBUG=true

The +asyncpg driver suffix tells SQLAlchemy to use the async PostgreSQL driver. (For setup of the actual database, see How to Connect PostgreSQL with Django — the SQL parts apply equally to FastAPI.)

Async engine, session factory, and base

# app/core/database.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from app.core.config import get_settings


settings = get_settings()

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_pre_ping=True,    # detect dead connections
    pool_size=10,
    max_overflow=20,
)

AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)


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

A few choices worth calling out:

expire_on_commit=False — without this, every committed object’s attributes get invalidated and re-fetched on next access. For async APIs (where you serialize the object after the session closes), this is what you want.
pool_pre_ping=True — pings the connection before handing it out. Costs ~1ms; saves you from “stale connection” errors when your DB restarts.
echo=settings.debug — logs every SQL statement when in debug mode. Invaluable in development.

The model

SQLAlchemy 2.0’s typed Mapped[] syntax is the modern way:

# app/models/task.py
from datetime import datetime
from typing import Optional

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

from app.core.database import Base


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[Optional[str]] = mapped_column(String, nullable=True)
    completed: Mapped[bool] = mapped_column(default=False, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
    )

Mapped[int] and Mapped[str] give you proper type checking — your IDE knows task.title is a str and yells if you assign an int to it.

Pydantic schemas

Pydantic schemas are the contract between your API and its clients. Keep them separate from ORM models.

# app/schemas/task.py
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field


class TaskBase(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str | None = None
    completed: bool = False


class TaskCreate(TaskBase):
    pass


class TaskUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    description: str | None = None
    completed: bool | None = None


class TaskRead(TaskBase):
    model_config = ConfigDict(from_attributes=True)

    id: int
    created_at: datetime
    updated_at: datetime

from_attributes=True lets Pydantic build a TaskRead from an SQLAlchemy Task object directly — no manual mapping.

Dependency injection: the DB session

This is the elegant part. We define a dependency that yields a session, and FastAPI injects it into every route that needs it.

# app/api/deps.py
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import AsyncSessionLocal


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

Each request gets a fresh session, automatically closed when the route returns. No global state, no leaked connections.

The routes

# app/api/routes/tasks.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.deps import get_db
from app.models.task import Task
from app.schemas.task import TaskCreate, TaskRead, TaskUpdate


router = APIRouter()


@router.post("/", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
async def create_task(payload: TaskCreate, db: AsyncSession = Depends(get_db)) -> Task:
    task = Task(**payload.model_dump())
    db.add(task)
    await db.commit()
    await db.refresh(task)
    return task


@router.get("/", response_model=list[TaskRead])
async def list_tasks(
    skip: int = 0,
    limit: int = 50,
    db: AsyncSession = Depends(get_db),
) -> list[Task]:
    result = await db.execute(
        select(Task).order_by(Task.created_at.desc()).offset(skip).limit(limit)
    )
    return list(result.scalars().all())


@router.get("/{task_id}", response_model=TaskRead)
async def get_task(task_id: int, db: AsyncSession = Depends(get_db)) -> Task:
    task = await db.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@router.patch("/{task_id}", response_model=TaskRead)
async def update_task(
    task_id: int, payload: TaskUpdate, db: AsyncSession = Depends(get_db)
) -> Task:
    task = await db.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    for field, value in payload.model_dump(exclude_unset=True).items():
        setattr(task, field, value)
    await db.commit()
    await db.refresh(task)
    return task


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: int, db: AsyncSession = Depends(get_db)) -> None:
    task = await db.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    await db.delete(task)
    await db.commit()

Each route is async, takes a typed body and a session, and returns the model directly — Pydantic handles serialization via from_attributes.

Wire up the app

# app/main.py
from fastapi import FastAPI

from app.api.routes import tasks
from app.core.config import get_settings


settings = get_settings()
app = FastAPI(title=settings.app_name, debug=settings.debug)


app.include_router(tasks.router, prefix="/tasks", tags=["tasks"])


@app.get("/health")
async def health() -> dict:
    return {"status": "ok"}

Run it:

fastapi dev app/main.py

Visit http://127.0.0.1:8000/docs for the auto-generated Swagger UI.

Alembic for migrations

alembic init alembic

Edit alembic.ini:

sqlalchemy.url = postgresql+asyncpg://tasksuser:tasksdbpass@localhost:5432/tasksdb

Edit alembic/env.py to use your models’ metadata so autogenerate works:

# alembic/env.py — relevant bits
from app.models.task import Task  # import all your models
from app.core.database import Base

target_metadata = Base.metadata

Generate your first migration:

alembic revision --autogenerate -m "create tasks table"
alembic upgrade head

--autogenerate reads your models, compares to the live DB schema, and writes a migration. Review it before applying — autogenerate isn’t perfect with constraint changes.

A note on N+1 in SQLAlchemy

SQLAlchemy has the same N+1 trap as the Django ORM. The fix is selectinload and joinedload:

from sqlalchemy.orm import selectinload

# Eager-load related objects in a single follow-up query
result = await db.execute(
    select(Task).options(selectinload(Task.tags))
)

Without it, every task.tags access fires a fresh query. With it, you get one query for tasks and one for all related tags. Same lesson as in Django ORM Deep Dive .

Testing this setup

A quick sanity check with httpx.AsyncClient:

# tests/test_tasks.py
import pytest
from httpx import AsyncClient, ASGITransport

from app.main import app


@pytest.mark.asyncio
async def test_create_and_get_task():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        response = await ac.post("/tasks/", json={"title": "Buy milk"})
        assert response.status_code == 201
        task_id = response.json()["id"]

        response = await ac.get(f"/tasks/{task_id}")
        assert response.status_code == 200
        assert response.json()["title"] == "Buy milk"

For real test isolation you’d point at a separate test database and roll back after each test — but that’s a topic for the testing FastAPI apps post.

Conclusion

This stack — FastAPI + SQLAlchemy 2.0 (async) + PostgreSQL + Alembic — is a serious modern Python backend. It scales to real production workloads and stays a pleasure to write. The patterns here (typed models, dependency-injected sessions, separate Pydantic schemas, Alembic for migrations) are the same patterns you’ll see in any well-run FastAPI codebase.

Want more on FastAPI?

Happy building!

Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .

What we’re building#

Project setup#

Configuration with pydantic-settings#

Async engine, session factory, and base#

The model#

Pydantic schemas#

Dependency injection: the DB session#

The routes#

Wire up the app#

Alembic for migrations#

A note on N+1 in SQLAlchemy#

Testing this setup#

Conclusion#

What we’re building

Project setup

Configuration with `pydantic-settings`

Async engine, session factory, and base

The model

Pydantic schemas

Dependency injection: the DB session

The routes

Wire up the app

Alembic for migrations

A note on N+1 in SQLAlchemy

Testing this setup

Conclusion