How to Build a REST API With FastAPI and SQLAlchemy in Python

FastAPI handles HTTP requests and response validation. SQLAlchemy handles the database. Together, they form one of the most common stacks for building Python REST APIs. This tutorial walks through building a complete CRUD API from scratch: setting up the database connection, defining ORM models, creating Pydantic schemas for request and response validation, writing endpoints for create, read, update, and delete operations, and wiring everything together with FastAPI's dependency injection system. By the end, you will have a working API backed by a real database that you can test directly from the browser using Swagger UI.

Project Setup

Install the dependencies and create the project structure:

pip install fastapi uvicorn sqlalchemy pydantic

fastapi-crud/
    app/
        __init__.py
        main.py          # FastAPI app instance, startup, router registration
        database.py      # Engine, session factory, get_db dependency
        models.py        # SQLAlchemy ORM models
        schemas.py       # Pydantic request/response schemas
        crud.py          # Database operation functions
        router.py        # API route handlers

This example uses SQLite for simplicity. The same code works with PostgreSQL, MySQL, or any other database supported by SQLAlchemy—just change the connection URL and install the appropriate driver (e.g., psycopg2-binary for PostgreSQL).

Step 1: Database Connection and Session

The database module creates the SQLAlchemy engine, configures the session factory, and defines a get_db dependency that provides a session to each request and closes it afterward.

# app/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},  # SQLite only
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

The get_db function is a FastAPI dependency that uses yield. Everything before the yield runs before the request handler. Everything after runs when the request finishes. The finally block guarantees the session closes even if the handler raises an exception. This prevents connection leaks.

Step 2: SQLAlchemy ORM Model

The ORM model defines the database table. Each class attribute maps to a column.

# app/models.py

from sqlalchemy import Column, Integer, String, Float, Boolean
from app.database import Base

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True, nullable=False)
    description = Column(String, nullable=True)
    price = Column(Float, nullable=False)
    is_available = Column(Boolean, default=True)

This creates an items table with five columns. The id column is an auto-incrementing primary key. The index=True on title adds a database index for faster lookups.

Step 3: Pydantic Schemas

Pydantic schemas define what the client sends and what the server returns. They are separate from the SQLAlchemy model because the API contract and the database schema serve different purposes.

# app/schemas.py

from pydantic import BaseModel, Field, ConfigDict

class ItemBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: str | None = None
    price: float = Field(..., gt=0)
    is_available: bool = True

class ItemCreate(ItemBase):
    """Schema for creating an item. Sent by the client."""
    pass

class ItemUpdate(BaseModel):
    """Schema for partial updates. All fields optional."""
    title: str | None = None
    description: str | None = None
    price: float | None = Field(default=None, gt=0)
    is_available: bool | None = None

class ItemOut(ItemBase):
    """Schema returned to the client. Includes database-generated fields."""
    id: int

    model_config = ConfigDict(from_attributes=True)

ItemCreate inherits required fields from ItemBase. ItemUpdate makes every field optional for partial updates. ItemOut adds the id field that comes from the database. The from_attributes=True setting tells Pydantic to read data from SQLAlchemy model attributes, allowing you to return ORM objects directly from your route handlers.

Step 4: CRUD Service Functions

These functions encapsulate the database operations. Route handlers call them instead of writing queries inline.

# app/crud.py

from sqlalchemy.orm import Session
from app import models, schemas

def get_items(db: Session, skip: int = 0, limit: int = 20):
    return db.query(models.Item).offset(skip).limit(limit).all()

def get_item(db: Session, item_id: int):
    return db.query(models.Item).filter(models.Item.id == item_id).first()

def create_item(db: Session, item: schemas.ItemCreate):
    db_item = models.Item(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def update_item(db: Session, item_id: int, item: schemas.ItemUpdate):
    db_item = get_item(db, item_id)
    if not db_item:
        return None
    update_data = item.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_item, key, value)
    db.commit()
    db.refresh(db_item)
    return db_item

def delete_item(db: Session, item_id: int):
    db_item = get_item(db, item_id)
    if not db_item:
        return False
    db.delete(db_item)
    db.commit()
    return True

The model_dump(exclude_unset=True) call in the update function is key. It returns only the fields the client included in the request, ignoring fields that were not sent. This enables true partial updates—the client can update just the price without touching the title or description.

Step 5: Route Handlers

The router defines the API endpoints. Each handler receives the database session through FastAPI's dependency injection and delegates the work to the CRUD functions.

# app/router.py

from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session

from app.database import get_db
from app import crud, schemas

router = APIRouter(prefix="/items", tags=["Items"])

DbSession = Annotated[Session, Depends(get_db)]

@router.get("/", response_model=list[schemas.ItemOut])
def list_items(
    db: DbSession,
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
):
    return crud.get_items(db, skip=skip, limit=limit)

@router.get("/{item_id}", response_model=schemas.ItemOut)
def read_item(item_id: int, db: DbSession):
    item = crud.get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.post("/", response_model=schemas.ItemOut, status_code=status.HTTP_201_CREATED)
def create_item(item: schemas.ItemCreate, db: DbSession):
    return crud.create_item(db, item)

@router.patch("/{item_id}", response_model=schemas.ItemOut)
def update_item(item_id: int, item: schemas.ItemUpdate, db: DbSession):
    updated = crud.update_item(db, item_id, item)
    if not updated:
        raise HTTPException(status_code=404, detail="Item not found")
    return updated

@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int, db: DbSession):
    if not crud.delete_item(db, item_id):
        raise HTTPException(status_code=404, detail="Item not found")

Notice how thin the handlers are. They accept input, call a CRUD function, handle the "not found" case, and return the result. There are no database queries, no business logic, and no manual JSON serialization in the handlers themselves.

Step 6: Wiring It Together in main.py

# app/main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.database import engine, Base
from app.router import router as items_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create tables on startup
    Base.metadata.create_all(bind=engine)
    yield

app = FastAPI(title="Items API", version="1.0.0", lifespan=lifespan)
app.include_router(items_router, prefix="/api/v1")

The lifespan context manager creates the database tables when the application starts. The router is included with an /api/v1 prefix for versioning. Start the server with:

uvicorn app.main:app --reload

Testing With Swagger UI

Open your browser and navigate to http://127.0.0.1:8000/docs. FastAPI generates interactive documentation from your route definitions and Pydantic schemas. You can create items, list them, update fields, and delete records directly from the browser without any external tools.

Try creating an item:

curl -X POST http://127.0.0.1:8000/api/v1/items/ \
  -H "Content-Type: application/json" \
  -d '{"title": "Mechanical Keyboard", "price": 89.99}'

Then update just the price with a partial update:

curl -X PATCH http://127.0.0.1:8000/api/v1/items/1 \
  -H "Content-Type: application/json" \
  -d '{"price": 79.99}'

The title and other fields remain unchanged because model_dump(exclude_unset=True) only applies the fields the client included.

Next Steps: Migrations With Alembic

The Base.metadata.create_all() call creates tables that do not exist, but it does not modify existing tables when your models change. For production applications, use Alembic to manage schema migrations.

pip install alembic
alembic init alembic

Configure alembic/env.py to import your Base metadata and database URL. Then generate and apply migrations:

# Generate a migration after changing your models
alembic revision --autogenerate -m "add items table"

# Apply the migration to the database
alembic upgrade head

Alembic tracks every schema change in version-controlled migration files. This gives you a reliable, repeatable process for evolving your database schema across development, staging, and production environments.

Frequently Asked Questions

How do I connect FastAPI to a database with SQLAlchemy?

Create a SQLAlchemy engine with your database URL, configure a session factory using sessionmaker, and define a get_db dependency function that yields a session and closes it when the request finishes. Inject this dependency into your route handlers using Depends(get_db). Each request gets its own session with guaranteed cleanup.

What is the difference between a SQLAlchemy model and a Pydantic schema in FastAPI?

A SQLAlchemy model defines a database table and its columns. A Pydantic schema defines the shape of data crossing your API boundary. Keep them separate so your API contract stays independent from your database structure. A creation schema might accept a plain-text password, while the database model stores a hash. A response schema might exclude internal fields entirely.

Should I use SQLAlchemy or SQLModel with FastAPI?

Both work well. SQLModel, created by the same author as FastAPI, combines SQLAlchemy and Pydantic into a single model class, which reduces duplication. Standard SQLAlchemy with separate Pydantic schemas gives you more explicit control and is the more widely used pattern in production codebases. For new projects, either choice is valid.

How do I handle database migrations with FastAPI and SQLAlchemy?

Use Alembic, the migration tool built for SQLAlchemy. Initialize it with alembic init, configure it to use your models and database URL, then run alembic revision --autogenerate to generate migration scripts from your model changes and alembic upgrade head to apply them.

Key Takeaways

1. Use the get_db dependency for session management: A yield dependency creates a session per request and closes it automatically. This prevents connection leaks and keeps your route handlers clean.
2. Keep SQLAlchemy models and Pydantic schemas separate: ORM models define database structure. Pydantic schemas define the API contract. Separating them gives you the flexibility to evolve each independently.
3. Put database logic in CRUD service functions: Route handlers should delegate to service functions rather than writing queries inline. This makes the logic reusable and testable.
4. Use model_dump(exclude_unset=True) for partial updates: This Pydantic v2 method returns only the fields the client included, enabling true PATCH behavior where omitted fields remain unchanged.
5. Use Alembic for migrations in production: Base.metadata.create_all() works for prototyping, but Alembic provides version-controlled, repeatable schema migrations that are essential for production deployments.

Building a REST API with FastAPI and SQLAlchemy follows a clear separation of concerns. The database module manages connections. ORM models define tables. Pydantic schemas define the API contract. CRUD functions encapsulate queries. Route handlers receive validated input, call service functions, and return structured responses. Each layer has a single job, which makes the codebase straightforward to navigate, test, and extend as your application grows.