SQLModel: the basics

SQLModel is a library for working with databases in FastAPI, combining SQLAlchemy and Pydantic.

Installation

pip install sqlmodel

Models

from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List
from datetime import datetime

class Project(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(max_length=200)
    created_at: datetime = Field(default_factory=datetime.utcnow)

    tasks: List["Task"] = Relationship(back_populates="project")

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(max_length=200, index=True)
    status: str = Field(default="todo")
    priority: int = Field(default=1)
    project_id: Optional[int] = Field(default=None, foreign_key="project.id")

    project: Optional[Project] = Relationship(back_populates="tasks")

Database connection

from sqlmodel import create_engine, Session, SQLModel

DATABASE_URL = "sqlite:///./db.sqlite3"
# For PostgreSQL:
# DATABASE_URL = "postgresql://user:pass@localhost/mydb"

engine = create_engine(DATABASE_URL, echo=True)  # echo=True logs SQL queries

def create_tables():
    SQLModel.metadata.create_all(engine)

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

CRUD operations

from sqlmodel import select

def create_task(session: Session, title: str, project_id: int) -> Task:
    task = Task(title=title, project_id=project_id)
    session.add(task)
    session.commit()
    session.refresh(task)  # refresh from DB (to get the id)
    return task

def get_task(session: Session, task_id: int) -> Task | None:
    return session.get(Task, task_id)

def list_tasks(session: Session, status: str | None = None) -> list[Task]:
    query = select(Task)
    if status:
        query = query.where(Task.status == status)
    return session.exec(query).all()

def update_task(session: Session, task: Task, **kwargs) -> Task:
    for key, value in kwargs.items():
        setattr(task, key, value)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

def delete_task(session: Session, task: Task):
    session.delete(task)
    session.commit()

Schemas without table=True

class TaskCreate(SQLModel):      # API-only schema, no table=True
    title: str
    priority: int = 1

class TaskResponse(SQLModel):
    id: int
    title: str
    status: str
    created_at: datetime
    model_config = {"from_attributes": True}