FASTAPI / 6. DATABASE WITH SQLMODEL
Database Integration with SQLModel
SQL databases with Pydantic-native ORM — SQLite, PostgreSQL, MySQL
EXPLANATION
SQLModel is the official companion library for FastAPI, created by the same author (Sebastián Ramírez). It merges Pydantic models and SQLAlchemy ORM into one class — your data model is your API model. The pattern: one class drives everything: • Define a class with table=True → it becomes a SQL table (SQLAlchemy) • Use the same class for request/response validation (Pydantic) • Create separate "Create" and "Read" variants for input vs output SQLModel under the hood: • Uses SQLAlchemy for the actual SQL generation and execution • Uses Pydantic for validation and serialization • Session is from SQLModel but wraps SQLAlchemy's Session • Supports SQLite (dev), PostgreSQL, MySQL (production) The session pattern with Depends(): • get_db() opens a Session, yields it, then closes it (finally) • FastAPI calls get_db() before the route, injects the session • After the route returns (or raises), the session closes • This guarantees no leaked connections CRUD operations: • Create → session.add(obj) then session.commit() then session.refresh(obj) • Read → session.get(Model, id) or session.exec(select(Model).where(...)) • Update → fetch object, mutate fields, session.add(obj), session.commit() • Delete → session.delete(obj), session.commit() Relationships: SQLModel supports one-to-many and many-to-many using SQLAlchemy's Relationship and FK columns. Use selectinload for eager loading to avoid N+1 queries.
ARCHITECTURE
Single class → multiple roles:
class Task(SQLModel, table=True): ← SQL table definition
id: int | None = Field(primary_key=True)
title: str
done: bool = False
│
┌──────┼──────┐
↓ ↓
SQLAlchemy ORM Pydantic Model
(creates table) (validates data)
Database session lifecycle:
┌──────────────────────────────────────┐
│ def get_db(): │
│ with Session(engine) as db: │
│ yield db ← route runs │
│ # auto-closed here │
└──────────────────────────────────────┘
SQL generated automatically:
session.exec(select(Task).where(Task.done == False))
→ SELECT * FROM task WHERE done = 0CODE