How To Use Python SQLModel for Database Models with FastAPI

Intermediate

If you’ve built a FastAPI application with a database backend, you’ve probably run into this: you define a Pydantic model for request/response validation, then a separate SQLAlchemy model for the database table, and then spend time keeping them in sync. Change one field and you have to update both. Add a validator and you need it in both places. SQLModel solves this by merging the two into a single class that works as both a Pydantic model and a SQLAlchemy ORM model.

SQLModel is built by the same developer as FastAPI (Sebastian Ramirez) and is designed to integrate seamlessly with FastAPI’s dependency injection and OpenAPI documentation. Under the hood it’s a thin layer over SQLAlchemy 2.0 and Pydantic v2, so you get the full power of both — type-checked models that also map to database tables, automatic schema generation, and Alembic migration support. You’ll need Python 3.7+ and you install it with pip install sqlmodel.

In this tutorial you’ll learn how to define SQLModel table models, create and query a database with the SQLAlchemy engine, relate tables with foreign keys, use SQLModel with FastAPI for a complete CRUD API, and handle optional fields correctly for create vs update operations. By the end you’ll have a working FastAPI + SQLite application that demonstrates the full SQLModel workflow.

Python SQLModel: Quick Example

Here’s a minimal SQLModel app that creates a table, inserts a row, and queries it back — all in under 20 lines:

# sqlmodel_quick.py
from sqlmodel import Field, Session, SQLModel, create_engine, select

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    power: str
    level: int = 1

engine = create_engine("sqlite:///heroes.db", echo=False)
SQLModel.metadata.create_all(engine)

# Insert
with Session(engine) as session:
    hero = Hero(name="Spider-Man", power="wall-crawling", level=5)
    session.add(hero)
    session.commit()
    session.refresh(hero)
    print(f"Created hero ID={hero.id}")

# Query
with Session(engine) as session:
    heroes = session.exec(select(Hero)).all()
    for h in heroes:
        print(f"{h.id}: {h.name} (level {h.level})")

Output:

Created hero ID=1
1: Spider-Man (level 5)

The key is table=True in the class definition — this tells SQLModel to register the class as a database table. Without it, the class is just a Pydantic model. The Field(primary_key=True) maps to SQLAlchemy’s primary key constraint, and default=None lets the database auto-assign the ID on insert. Notice how session.exec(select(Hero)) returns fully typed Hero objects — no manual type conversion needed.

The sections below cover relationships, FastAPI integration, and the create/update pattern in depth.

What Is SQLModel and Why Use It?

SQLModel is a Python ORM library that unifies SQLAlchemy and Pydantic into a single model definition. A traditional FastAPI + SQLAlchemy stack requires two parallel class hierarchies: one for database tables (SQLAlchemy declarative base) and one for API schemas (Pydantic BaseModel). SQLModel replaces both with a single SQLModel base class that is simultaneously a Pydantic model and a SQLAlchemy ORM model.

Feature	SQLAlchemy alone	Pydantic alone	SQLModel
Database ORM	Yes	No	Yes
Runtime type validation	No	Yes	Yes
FastAPI schema generation	No	Yes	Yes
Single model definition	No	No	Yes
Alembic migration support	Yes	No	Yes
Async support	Yes (asyncio)	N/A	Yes (AsyncSession)

The trade-off is that SQLModel adds an abstraction layer, so some advanced SQLAlchemy patterns require dropping down to the SQLAlchemy API. For most CRUD applications, SQLModel’s higher-level API is exactly the right level of abstraction.

Defining Table and Non-Table Models

The key design choice in SQLModel is the table=True flag. Models with this flag map to database tables; models without it are pure Pydantic schemas used for API input/output validation. This lets you define separate schemas for “create” (no ID field), “read” (ID included), and “update” (all fields optional) while sharing the field definitions from the table model.

# sqlmodel_models.py
from sqlmodel import Field, SQLModel
from typing import Optional

# --- Database table model ---
class Item(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)          # creates DB index
    description: Optional[str] = None
    price: float
    in_stock: bool = True

# --- API schemas (no table=True) ---
class ItemCreate(SQLModel):
    """Fields required to create an item. No id -- DB assigns it."""
    name: str
    description: Optional[str] = None
    price: float
    in_stock: bool = True

class ItemRead(SQLModel):
    """Fields returned in API responses. id is always present."""
    id: int
    name: str
    description: Optional[str]
    price: float
    in_stock: bool

class ItemUpdate(SQLModel):
    """All fields optional for PATCH updates."""
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    in_stock: Optional[bool] = None

# Validate that ItemCreate works as expected
item = ItemCreate(name="Widget", price=9.99)
print(item.model_dump())

# Invalid data raises a ValidationError
try:
    bad = ItemCreate(name="Widget", price="not-a-number")
except Exception as e:
    print(f"Validation error: {type(e).__name__}")

Output:

{'name': 'Widget', 'description': None, 'price': 9.99, 'in_stock': True}
Validation error: ValidationError

Using separate ItemCreate, ItemRead, and ItemUpdate schemas alongside the Item table model is the standard SQLModel pattern for FastAPI. The ItemCreate schema is what the API accepts; the ItemRead schema is what the API returns; the Item table model is what the database uses. FastAPI’s response model validation uses ItemRead to strip any fields that shouldn’t appear in the response.

CRUD Operations with Session

SQLModel uses the same Session pattern as SQLAlchemy 2.0. Every database operation runs inside a session context manager. The select() function builds type-safe queries and session.exec() executes them, returning strongly-typed results rather than raw row tuples.

# sqlmodel_crud.py
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing import Optional
import os

class Item(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    price: float
    in_stock: bool = True

DB_PATH = "items.db"
engine = create_engine(f"sqlite:///{DB_PATH}", echo=False)
SQLModel.metadata.create_all(engine)

def create_item(name: str, price: float) -> Item:
    with Session(engine) as session:
        item = Item(name=name, price=price)
        session.add(item)
        session.commit()
        session.refresh(item)   # load auto-assigned id
        return item

def get_item(item_id: int) -> Optional[Item]:
    with Session(engine) as session:
        return session.get(Item, item_id)

def list_items(min_price: float = 0.0) -> list[Item]:
    with Session(engine) as session:
        stmt = select(Item).where(Item.price >= min_price).order_by(Item.name)
        return session.exec(stmt).all()

def update_item(item_id: int, new_price: float) -> Optional[Item]:
    with Session(engine) as session:
        item = session.get(Item, item_id)
        if not item:
            return None
        item.price = new_price
        session.add(item)
        session.commit()
        session.refresh(item)
        return item

def delete_item(item_id: int) -> bool:
    with Session(engine) as session:
        item = session.get(Item, item_id)
        if not item:
            return False
        session.delete(item)
        session.commit()
        return True

# Demo
widget = create_item("Widget", 9.99)
gadget = create_item("Gadget", 24.99)
print(f"Created: {widget.id} {widget.name}, {gadget.id} {gadget.name}")

items = list_items(min_price=10.0)
print(f"Items >= $10: {[i.name for i in items]}")

updated = update_item(widget.id, 12.99)
print(f"Updated widget price: ${updated.price}")

deleted = delete_item(gadget.id)
print(f"Deleted gadget: {deleted}")
print(f"Remaining items: {len(list_items())}")

# Cleanup
os.remove(DB_PATH)

Output:

Created: 1 Widget, 2 Gadget
Items >= $10: ['Gadget']
Updated widget price: $12.99
Deleted gadget: True
Remaining items: 1

session.get(Item, item_id) is the efficient single-row lookup by primary key — it hits the identity cache first and only queries the database if the object isn’t already loaded. Always call session.refresh(item) after commit() to reload auto-generated values like id or database-side defaults. Without refresh(), accessing item.id after a commit can raise a DetachedInstanceError.

Defining Relationships with Foreign Keys

SQLModel supports table relationships using Relationship and standard SQLAlchemy foreign keys. Relationships let you navigate between related objects in Python without writing manual join queries.

# sqlmodel_relationships.py
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
from typing import Optional, List

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    heroes: List["Hero"] = Relationship(back_populates="team")

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    team_id: Optional[int] = Field(default=None, foreign_key="team.id")
    team: Optional[Team] = Relationship(back_populates="heroes")

engine = create_engine("sqlite:///teams.db", echo=False)
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    # Create team and heroes together
    avengers = Team(name="Avengers")
    session.add(avengers)
    session.commit()
    session.refresh(avengers)

    iron_man = Hero(name="Iron Man", team_id=avengers.id)
    thor = Hero(name="Thor", team_id=avengers.id)
    session.add_all([iron_man, thor])
    session.commit()

    # Query heroes with their team
    stmt = select(Hero).where(Hero.team_id == avengers.id)
    members = session.exec(stmt).all()
    print(f"{avengers.name} has {len(members)} heroes:")
    for h in members:
        print(f"  - {h.name}")

import os; os.remove("teams.db")

Output:

Avengers has 2 heroes:
  - Iron Man
  - Thor

The back_populates parameter creates a two-way relationship: team.heroes gives you the list of heroes in a team, and hero.team gives you the hero’s team. SQLModel uses lazy loading by default — the related objects are only fetched when you actually access the relationship attribute. For performance-critical applications, use selectin loading or explicit joins to avoid N+1 query problems.

FastAPI Integration: Complete CRUD API

SQLModel’s real strength shows when combined with FastAPI. The same model classes serve as both database tables and FastAPI request/response schemas, and FastAPI automatically generates OpenAPI documentation for them.

# sqlmodel_fastapi.py
from fastapi import FastAPI, HTTPException, Depends
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing import Optional, List

# --- Models ---
class HeroBase(SQLModel):
    name: str
    power: str
    level: int = 1

class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class HeroCreate(HeroBase):
    pass  # same fields as base, no id

class HeroRead(HeroBase):
    id: int

class HeroUpdate(SQLModel):
    name: Optional[str] = None
    power: Optional[str] = None
    level: Optional[int] = None

# --- Database setup ---
engine = create_engine("sqlite:///heroes.db", echo=False)
SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

# --- FastAPI app ---
app = FastAPI(title="Heroes API")

@app.post("/heroes/", response_model=HeroRead, status_code=201)
def create_hero(hero: HeroCreate, session: Session = Depends(get_session)):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

@app.get("/heroes/", response_model=List[HeroRead])
def read_heroes(skip: int = 0, limit: int = 10, session: Session = Depends(get_session)):
    heroes = session.exec(select(Hero).offset(skip).limit(limit)).all()
    return heroes

@app.get("/heroes/{hero_id}", response_model=HeroRead)
def read_hero(hero_id: int, session: Session = Depends(get_session)):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

@app.patch("/heroes/{hero_id}", response_model=HeroRead)
def update_hero(hero_id: int, hero_update: HeroUpdate, session: Session = Depends(get_session)):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    update_data = hero_update.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(hero, field, value)
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: Session = Depends(get_session)):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

Run with:

pip install fastapi uvicorn sqlmodel
uvicorn sqlmodel_fastapi:app --reload
# Docs at http://127.0.0.1:8000/docs

The model_validate(hero) call converts a HeroCreate Pydantic model into a Hero table model — SQLModel inherits Pydantic’s model_validate for this. The Depends(get_session) pattern injects a fresh session per request and automatically closes it when the request completes. hero_update.model_dump(exclude_unset=True) is the key to PATCH support: it only returns fields that were actually sent in the request, not fields that defaulted to None.

Real-Life Example: Task Manager API

A complete task manager with users, tasks, and status tracking — demonstrating relationships, filtering, and pagination in a realistic SQLModel + FastAPI app.

# task_manager.py
from fastapi import FastAPI, HTTPException, Depends, Query
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
from typing import Optional, List
from datetime import datetime
from enum import Enum

class TaskStatus(str, Enum):
    todo = "todo"
    in_progress = "in_progress"
    done = "done"

class UserBase(SQLModel):
    username: str
    email: str

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    tasks: List["Task"] = Relationship(back_populates="owner")

class UserCreate(UserBase):
    pass

class UserRead(UserBase):
    id: int

class TaskBase(SQLModel):
    title: str
    description: Optional[str] = None
    status: TaskStatus = TaskStatus.todo

class Task(TaskBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    owner_id: Optional[int] = Field(default=None, foreign_key="user.id")
    owner: Optional[User] = Relationship(back_populates="tasks")

class TaskCreate(TaskBase):
    owner_id: int

class TaskRead(TaskBase):
    id: int
    created_at: datetime
    owner_id: int

class TaskUpdate(SQLModel):
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[TaskStatus] = None

engine = create_engine("sqlite:///tasks.db", echo=False)
SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

app = FastAPI(title="Task Manager API")

@app.post("/users/", response_model=UserRead, status_code=201)
def create_user(user: UserCreate, session: Session = Depends(get_session)):
    db_user = User.model_validate(user)
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user

@app.post("/tasks/", response_model=TaskRead, status_code=201)
def create_task(task: TaskCreate, session: Session = Depends(get_session)):
    db_task = Task.model_validate(task)
    session.add(db_task)
    session.commit()
    session.refresh(db_task)
    return db_task

@app.get("/tasks/", response_model=List[TaskRead])
def list_tasks(
    status: Optional[TaskStatus] = None,
    owner_id: Optional[int] = None,
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=10, le=100),
    session: Session = Depends(get_session)
):
    stmt = select(Task)
    if status:
        stmt = stmt.where(Task.status == status)
    if owner_id:
        stmt = stmt.where(Task.owner_id == owner_id)
    stmt = stmt.offset(skip).limit(limit)
    return session.exec(stmt).all()

@app.patch("/tasks/{task_id}", response_model=TaskRead)
def update_task(task_id: int, update: TaskUpdate, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    for field, value in update.model_dump(exclude_unset=True).items():
        setattr(task, field, value)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

Run with:

uvicorn task_manager:app --reload
# POST /users/  -> creates user
# POST /tasks/  -> creates task for that user
# GET /tasks/?status=todo&owner_id=1  -> filtered task list

The TaskStatus enum ensures the API only accepts valid status values and documents the allowed choices in the OpenAPI schema automatically. The filtering pattern in list_tasks is composable — you can add more .where()` clauses without rewriting the query. Extend this example by adding authentication with FastAPI's OAuth2 dependency, Alembic migrations for schema changes, or async sessions with AsyncSession for high-concurrency workloads.

Frequently Asked Questions

Is SQLModel a replacement for SQLAlchemy?

SQLModel is built on top of SQLAlchemy, not a replacement for it. It provides a higher-level API that integrates with Pydantic, but all SQLAlchemy features are still available. You can mix SQLModel's select() with SQLAlchemy's advanced query constructs, use Alembic for migrations (SQLModel.metadata is a standard SQLAlchemy MetaData object), and access the underlying engine and session objects directly. When SQLModel's API is insufficient, drop down to SQLAlchemy -- they're fully compatible.

Does SQLModel support async databases?

Yes. Use create_async_engine from SQLAlchemy and AsyncSession from sqlmodel.ext.asyncio (or directly from sqlalchemy.ext.asyncio). The query syntax is identical, but you await session.exec() instead of calling it synchronously. For databases, use an async driver like aiosqlite for SQLite or asyncpg for PostgreSQL. The async pattern is recommended for high-concurrency FastAPI applications where database calls should not block the event loop.

How do I handle database migrations with SQLModel?

Use Alembic -- the standard SQLAlchemy migration tool. Run alembic init alembic to create the migration environment, then set target_metadata = SQLModel.metadata in alembic/env.py. Run alembic revision --autogenerate -m "add column" to auto-generate a migration from your model changes, and alembic upgrade head to apply it. Never use SQLModel.metadata.create_all() in production -- that only creates tables that don't exist and won't handle column additions, renames, or deletions.

Why use Optional[str] = None for non-required fields?

SQLModel inherits Pydantic's field semantics: a field with Optional[str] = None is optional in both the Python model and the database column (nullable). This creates a nullable column in the database. If you want a column that has a database default but is required in Python, use Field(default=None, sa_column_kwargs={"server_default": "value"}). The exclude_unset=True pattern in PATCH endpoints relies on distinguishing "field not sent" (not in model dump) from "field sent as None" (in dump as None).

Can SQLModel work with PostgreSQL and MySQL?

Yes -- SQLModel uses SQLAlchemy's database abstraction layer, so it works with any database SQLAlchemy supports: PostgreSQL (use psycopg2 or asyncpg), MySQL/MariaDB (use mysqlclient or aiomysql), SQLite (built-in), and more. Change the connection URL in create_engine(): "postgresql://user:pass@localhost/dbname" for PostgreSQL. Field types like str and int map to the appropriate column types for each database automatically.

Conclusion

SQLModel eliminates the duplication between SQLAlchemy ORM models and Pydantic schemas by merging them into a single class. The core workflow is: define a table=True model for the database, define non-table models for API input/output, use Session.exec(select(Model))` for queries, and use model_dump(exclude_unset=True) for PATCH operations. The FastAPI integration is seamless because both libraries share the same author and design philosophy.

The task manager example above is a solid foundation for a real application. Next steps: add Alembic for migrations, switch to AsyncSession for async support, add authentication with FastAPI's Depends system, and add indexes to frequently-filtered columns with Field(index=True).

Official documentation: https://sqlmodel.tiangolo.com/

Related Articles

• How To Build a REST API with FastAPI in Python
• How To Use Pydantic V2 for Data Validation in Python
• How To Use SQLAlchemy 2.0 ORM with Python