FastAPI CRUD API with Pydantic, SQLAlchemy and SQLite

A CRUD API (create, read, update, delete) is the "hello world" of backend work, and it's also most of what real business apps do. FastAPI makes it pleasant: you describe your data with Python type hints, and you get validation, error messages and interactive documentation for free.

We'll build a small product API in a single file, then test every endpoint. All the code below runs as-is.

Why FastAPI?

When I built a small chess server prototype, plain Django gave roughly a 1.5-second delay per move. Bringing in FastAPI, then WebSockets, Redis and separate services, is what made play feel instant. You don't need any of that for a CRUD API, but it's why I reach for FastAPI whenever an API needs to be quick to build and quick to respond.

The bigger reason for beginners: FastAPI reads your type hints and does the boring work. (If hints still feel like decoration, read what type hints actually do at runtime.)

The plan: three layers

Think of a restaurant:

The kitchen is the database. SQLAlchemy models describe what's stored on the shelves.
The menu is the API contract. Pydantic schemas say what customers can order and what they get back.
The waiter is the endpoint, taking the order, talking to the kitchen and bringing the plate.

The menu doesn't list the kitchen's internal codes, and the kitchen doesn't care how the menu is printed. Keeping them separate is what keeps the app easy to change.

Setup

pip install "fastapi[standard]" sqlalchemy

Create main.py. Each block below goes into that file, in order.

1. The database (the kitchen)

from contextlib import asynccontextmanager
from typing import Annotated
 
from fastapi import Depends, FastAPI, HTTPException, Query, status
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
 
engine = create_engine("sqlite:///shop.db", connect_args={"check_same_thread": False})
 
 
class Base(DeclarativeBase):
    pass
 
 
class Product(Base):
    __tablename__ = "products"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    price: Mapped[float]
    in_stock: Mapped[bool] = mapped_column(default=True)

This is SQLAlchemy 2.0 style. Mapped[int] tells SQLAlchemy the column type from the hint, so price: Mapped[float] needs no extra arguments.

check_same_thread=False is a SQLite detail: FastAPI may handle a request on a different thread than the one that opened the connection, and SQLite refuses that by default.

2. One session per request

def get_session():
    with Session(engine) as session:
        yield session
 
 
SessionDep = Annotated[Session, Depends(get_session)]

This is a dependency. Any endpoint that asks for SessionDep gets a fresh database session, and the with block closes it after the response, even if the endpoint crashes. No leaked connections.

class ProductCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)
    in_stock: bool = True
 
 
class ProductUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=100)
    price: float | None = Field(default=None, gt=0)
    in_stock: bool | None = None
 
 
class ProductRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
 
    id: int
    name: str
    price: float
    in_stock: bool

Three schemas, three jobs:

ProductCreate: what a client must send to create a product. No id; the database assigns that. gt=0 means price must be greater than zero.
ProductUpdate: every field optional, because a client may change just one.
ProductRead: what we send back. from_attributes=True lets Pydantic read directly from a SQLAlchemy object.

4. Create the table on startup

@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(engine)
    yield
 
 
app = FastAPI(title="Shop API", lifespan=lifespan)

The code before yield runs once when the server starts. create_all makes any missing tables. (In a real project you'd use migrations with Alembic instead, so you can change tables safely later.)

5. The endpoints (the waiters)

Create

@app.post("/products", response_model=ProductRead, status_code=status.HTTP_201_CREATED)
def create_product(data: ProductCreate, session: SessionDep):
    product = Product(**data.model_dump())
    session.add(product)
    session.commit()
    session.refresh(product)
    return product

FastAPI validates the JSON body against ProductCreate before your function runs. If it's invalid, your code never executes. refresh reloads the row so the new id is filled in. 201 Created is the correct status for a new resource.

Read many, with pagination

@app.get("/products", response_model=list[ProductRead])
def list_products(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 20,
):
    stmt = select(Product).order_by(Product.id).offset(offset).limit(limit)
    return session.scalars(stmt).all()

offset and limit become query parameters: /products?offset=20&limit=20. The le=100 cap stops a client from requesting a million rows in one go.

Read one, update, delete

def get_or_404(session: Session, product_id: int) -> Product:
    product = session.get(Product, product_id)
    if product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    return product
 
 
@app.get("/products/{product_id}", response_model=ProductRead)
def read_product(product_id: int, session: SessionDep):
    return get_or_404(session, product_id)
 
 
@app.patch("/products/{product_id}", response_model=ProductRead)
def update_product(product_id: int, data: ProductUpdate, session: SessionDep):
    product = get_or_404(session, product_id)
    for field, value in data.model_dump(exclude_unset=True).items():
        setattr(product, field, value)
    session.commit()
    session.refresh(product)
    return product
 
 
@app.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_product(product_id: int, session: SessionDep):
    product = get_or_404(session, product_id)
    session.delete(product)
    session.commit()

The important line is model_dump(exclude_unset=True). It returns only the fields the client actually sent. Send {"price": 999} and only price changes. Without exclude_unset, every missing field would come through as None and wipe your data.

get_or_404 keeps the "not found" logic in one place instead of copying it into three endpoints.

Run it

fastapi dev main.py

Open http://127.0.0.1:8000/docs. FastAPI built interactive documentation from your hints and schemas: every endpoint, every field and every rule, with a "Try it out" button. That page alone saves hours of back-and-forth with frontend developers.

Test it

Create test_main.py (install pytest with pip install pytest):

from fastapi.testclient import TestClient
from main import app
 
 
def test_crud():
    with TestClient(app) as client:
        r = client.post("/products", json={"name": "Desk lamp", "price": 1250})
        assert r.status_code == 201
        product_id = r.json()["id"]
 
        r = client.patch(f"/products/{product_id}", json={"price": 999})
        assert r.json()["price"] == 999
 
        r = client.post("/products", json={"name": "", "price": -5})
        assert r.status_code == 422
 
        assert client.delete(f"/products/{product_id}").status_code == 204
        assert client.get(f"/products/{product_id}").status_code == 404

Run pytest. The invalid product returns 422 with clear messages, and nothing in your code had to check it:

String should have at least 1 character
Input should be greater than 0

The with TestClient(app) block matters: it runs the lifespan startup, so the table exists before the first request.

What goes wrong

Using one model for everything. Returning SQLAlchemy objects without a response_model can leak fields you never meant to expose, like an internal cost price or a password hash.
async def with a synchronous database. It blocks the whole server while the query runs. With the standard SQLite driver, use plain def as we did, and FastAPI runs it in a thread pool.
PATCH without exclude_unset. Partial updates silently erase fields.
create_all in production. It never changes existing tables. Use Alembic migrations once the app is real.

Where to go next

Add authentication (JWT tokens are common; the JWT decoder helps you see what's inside one), move to PostgreSQL by changing the connection URL, and split the file into models.py, schemas.py and routers as it grows. When you're debugging responses, the JSON formatter makes them readable.

In one line: SQLAlchemy models describe storage, Pydantic schemas describe the API, a session dependency handles connections, and FastAPI turns your type hints into validation and docs.

Build a CRUD API with FastAPI, Pydantic, SQLAlchemy and SQLite