Project 9: Database ORM Integration
Build a data layer using SQLModel (Pydantic + SQLAlchemy) that shares models between API validation and database operations, with migrations and relationship handling.
Learning Objectives
By completing this project, you will:
- Understand SQLModel’s unified architecture - How it combines Pydantic and SQLAlchemy into a single model
- Master table=True vs table=False patterns - When to create database tables vs validation-only models
- Handle relationships correctly - Foreign keys, one-to-many, many-to-many with proper loading strategies
- Implement the repository pattern - Clean separation between domain logic and data access
- Manage database migrations with Alembic - Schema evolution without data loss
- Use async database operations - Non-blocking database access for high-performance APIs
Deep Theoretical Foundation
Why SQLModel? The Unification Story
Before SQLModel, developers faced a frustrating duplication problem:
WITHOUT SQLModel (The Pain):
┌────────────────────────────────────────────────────────────────────────────┐
│ │
│ # Pydantic model for API validation │
│ class UserCreate(BaseModel): │
│ email: str │
│ name: str │
│ password: str │
│ │
│ # SQLAlchemy model for database │
│ class User(Base): │
│ __tablename__ = "users" │
│ id = Column(Integer, primary_key=True) │
│ email = Column(String, unique=True) ◄── Duplicated! │
│ name = Column(String) ◄── Duplicated! │
│ hashed_password = Column(String) │
│ │
│ # You need ANOTHER model for responses │
│ class UserResponse(BaseModel): │
│ id: int │
│ email: str ◄── Duplicated again! │
│ name: str ◄── Duplicated again! │
│ │
└────────────────────────────────────────────────────────────────────────────┘
WITH SQLModel (The Solution):
┌────────────────────────────────────────────────────────────────────────────┐
│ │
│ class UserBase(SQLModel): │
│ email: EmailStr # Shared fields │
│ name: str │
│ │
│ class User(UserBase, table=True): # Database table │
│ id: Optional[int] = Field(primary_key=True) │
│ hashed_password: str │
│ │
│ class UserCreate(UserBase): # API input │
│ password: str │
│ │
│ class UserRead(UserBase): # API output │
│ id: int │
│ │
│ # Same field definitions, different purposes! │
│ │
└────────────────────────────────────────────────────────────────────────────┘
SQLModel was created by Sebastián Ramírez (creator of FastAPI) to solve this exact problem. It inherits from both Pydantic’s BaseModel and SQLAlchemy’s declarative base.
Understanding table=True vs table=False
The table parameter is the key that determines whether a SQLModel class creates a database table:
from sqlmodel import SQLModel, Field
from typing import Optional
# table=False (default): Pure Pydantic validation model
class UserBase(SQLModel):
"""No database table created. Used for validation only."""
email: str
name: str
# This class is essentially a Pydantic BaseModel
# table=True: Creates a database table
class User(UserBase, table=True):
"""Database table 'user' is created with these columns."""
id: Optional[int] = Field(default=None, primary_key=True)
hashed_password: str
# This class maps to a real database table
What happens internally:
When table=False (default):
┌─────────────────────────────────────────────────────────┐
│ class UserBase(SQLModel): │
│ │
│ Pydantic Features: │
│ ✓ Type validation │
│ ✓ JSON serialization │
│ ✓ Schema generation │
│ │
│ SQLAlchemy Features: │
│ ✗ No __tablename__ │
│ ✗ No Column() mappings │
│ ✗ No relationship() support │
│ ✗ Not registered with MetaData │
└─────────────────────────────────────────────────────────┘
When table=True:
┌─────────────────────────────────────────────────────────┐
│ class User(SQLModel, table=True): │
│ │
│ Pydantic Features: │
│ ✓ Type validation │
│ ✓ JSON serialization │
│ ✓ Schema generation │
│ │
│ SQLAlchemy Features: │
│ ✓ __tablename__ = "user" (auto-generated) │
│ ✓ id → Column(Integer, primary_key=True) │
│ ✓ email → Column(String) │
│ ✓ Registered with MetaData │
│ ✓ Can use Relationship() │
└─────────────────────────────────────────────────────────┘
The Model Inheritance Pattern
The recommended pattern for SQLModel projects:
┌─────────────────────┐
│ UserBase │
│ (table=False) │
│ │
│ - email: str │
│ - name: str │
│ - is_active: bool │
└──────────┬──────────┘
│
┌───────────────────┼───────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ UserCreate │ │ User │ │ UserRead │
│ (table=False) │ │ (table=True) │ │ (table=False) │
│ │ │ │ │ │
│ Inherits base + │ │ Inherits base + │ │ Inherits base + │
│ + password: str │ │ + id: int │ │ + id: int │
│ │ │ + hashed_pwd │ │ + created_at │
│ │ │ + created_at │ │ │
│ Used for: │ │ + orders: list │ │ Used for: │
│ POST /users │ │ │ │ GET /users/{id} │
└─────────────────┘ │ Database table │ └─────────────────┘
└─────────────────┘
Relationship Handling and Foreign Keys
SQLModel supports SQLAlchemy-style relationships:
from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List
class Team(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
# One-to-many: A team has many heroes
heroes: List["Hero"] = Relationship(back_populates="team")
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
# Foreign key column
team_id: Optional[int] = Field(default=None, foreign_key="team.id")
# Relationship back to Team
team: Optional[Team] = Relationship(back_populates="heroes")
Relationship types:
ONE-TO-MANY (User has many Orders):
┌─────────────┐ ┌─────────────┐
│ User │ 1 N │ Order │
│─────────────│◄────────│─────────────│
│ id (PK) │ │ id (PK) │
│ name │ │ user_id (FK)│
│ │ │ total │
└─────────────┘ └─────────────┘
class User(SQLModel, table=True):
id: Optional[int] = Field(primary_key=True)
orders: List["Order"] = Relationship(back_populates="user")
class Order(SQLModel, table=True):
id: Optional[int] = Field(primary_key=True)
user_id: int = Field(foreign_key="user.id")
user: Optional[User] = Relationship(back_populates="orders")
MANY-TO-MANY (Users have many Roles, Roles have many Users):
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ User │ N M │ UserRole │ M N │ Role │
│─────────────│◄────────│─────────────│────────►│─────────────│
│ id (PK) │ │ user_id (FK)│ │ id (PK) │
│ name │ │ role_id (FK)│ │ name │
└─────────────┘ └─────────────┘ └─────────────┘
class UserRoleLink(SQLModel, table=True):
user_id: int = Field(foreign_key="user.id", primary_key=True)
role_id: int = Field(foreign_key="role.id", primary_key=True)
class User(SQLModel, table=True):
id: Optional[int] = Field(primary_key=True)
roles: List["Role"] = Relationship(
back_populates="users",
link_model=UserRoleLink
)
class Role(SQLModel, table=True):
id: Optional[int] = Field(primary_key=True)
users: List["User"] = Relationship(
back_populates="roles",
link_model=UserRoleLink
)
The Repository Pattern
The repository pattern abstracts database operations behind a clean interface:
┌─────────────────────────────────────────────────────────────────────────┐
│ API Layer │
│ │
│ @app.post("/users") │
│ async def create_user(user: UserCreate): │
│ return user_repo.create(user) │
│ │
└─────────────────────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ Repository Layer │
│ │
│ class UserRepository: │
│ def create(self, data: UserCreate) -> User: ... │
│ def get_by_id(self, id: int) -> Optional[User]: ... │
│ def list(self, skip: int, limit: int) -> List[User]: ... │
│ def update(self, id: int, data: UserUpdate) -> User: ... │
│ def delete(self, id: int) -> bool: ... │
│ │
│ Benefits: │
│ - Testable (can mock the repository) │
│ - Swappable (can change database without changing API) │
│ - Clean (business logic doesn't know about SQL) │
│ │
└─────────────────────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ Database Layer │
│ │
│ SQLModel Session │
│ - session.add(user) │
│ - session.commit() │
│ - session.exec(select(User).where(...)) │
│ │
└─────────────────────────────────────────────────────────────────────────┘
Database Migrations with Alembic
Alembic manages schema changes over time:
Initial State: After Migration:
┌─────────────────────┐ ┌─────────────────────┐
│ users │ │ users │
│─────────────────────│ ──► │─────────────────────│
│ id: INTEGER (PK) │ │ id: INTEGER (PK) │
│ email: VARCHAR │ │ email: VARCHAR │
│ name: VARCHAR │ │ name: VARCHAR │
│ │ │ created_at: DATETIME│ ◄── NEW!
│ │ │ is_active: BOOLEAN │ ◄── NEW!
└─────────────────────┘ └─────────────────────┘
Migration file (versions/001_add_user_fields.py):
def upgrade():
op.add_column('users', sa.Column('created_at', sa.DateTime()))
op.add_column('users', sa.Column('is_active', sa.Boolean(), default=True))
def downgrade():
op.drop_column('users', 'created_at')
op.drop_column('users', 'is_active')
Alembic workflow:
# Initialize Alembic
alembic init alembic
# Generate migration from model changes
alembic revision --autogenerate -m "Add user fields"
# Apply migrations
alembic upgrade head
# Rollback
alembic downgrade -1
# View migration history
alembic history
Async Database Operations
For high-performance APIs, use async database operations:
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
# Async engine
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL, echo=True)
# Async session factory
async_session = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False
)
# Usage in FastAPI
async def get_session() -> AsyncSession:
async with async_session() as session:
yield session
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
session: AsyncSession = Depends(get_session)
):
result = await session.exec(
select(User).where(User.id == user_id)
)
user = result.first()
if not user:
raise HTTPException(404)
return user
Sync vs Async comparison:
SYNCHRONOUS (Blocking):
┌─────────────────────────────────────────────────────────────┐
│ Request 1 ████████████████████████░░░░░░░░░░░░░░░░░░░░░ │
│ Request 2 ░░░░░░░░░░░░░░░░░░░░░░░░████████████████████░░ │
│ Request 3 ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░████ │
│ │
│ Each request blocks until database responds │
└─────────────────────────────────────────────────────────────┘
ASYNCHRONOUS (Non-blocking):
┌─────────────────────────────────────────────────────────────┐
│ Request 1 ████░░░░░░░░░░████░░░░░░░░░░████ │
│ Request 2 ░░░░████░░░░░░░░░░████░░░░░░░░░░████ │
│ Request 3 ░░░░░░░░████░░░░░░░░░░████░░░░░░░░░░████ │
│ │
│ Requests interleave during I/O waits │
│ Same time, more throughput! │
└─────────────────────────────────────────────────────────────┘
from_attributes for ORM Compatibility
When returning ORM objects from FastAPI, you need from_attributes:
from pydantic import ConfigDict
from sqlmodel import SQLModel, Field
from typing import Optional
class UserRead(SQLModel):
"""Response model that can read from ORM objects."""
id: int
email: str
name: str
# This allows Pydantic to read from SQLAlchemy objects
model_config = ConfigDict(from_attributes=True)
# Now this works:
@app.get("/users/{user_id}", response_model=UserRead)
async def get_user(user_id: int, session: Session = Depends(get_session)):
user = session.get(User, user_id) # Returns SQLModel table instance
return user # Automatically converted to UserRead
How from_attributes works:
Without from_attributes=True:
┌─────────────────────────────────────────────────────────────┐
│ │
│ User(table=True) instance │
│ ├── id = 1 │
│ ├── email = "john@example.com" │
│ └── name = "John" │
│ │
│ ▼ Try to create UserRead │
│ │
│ ERROR: Input should be a valid dictionary │
│ (Pydantic expects dict, not ORM object) │
│ │
└─────────────────────────────────────────────────────────────┘
With from_attributes=True:
┌─────────────────────────────────────────────────────────────┐
│ │
│ User(table=True) instance │
│ ├── id = 1 │
│ ├── email = "john@example.com" │
│ └── name = "John" │
│ │
│ ▼ Pydantic reads attributes │
│ │
│ UserRead( │
│ id=getattr(user, 'id'), │
│ email=getattr(user, 'email'), │
│ name=getattr(user, 'name') │
│ ) │
│ │
│ SUCCESS: UserRead(id=1, email="john@example.com", ...) │
│ │
└─────────────────────────────────────────────────────────────┘
Project Specification
Functional Requirements
Build a complete e-commerce data layer that demonstrates:
- User management with authentication fields
- Product catalog with categories and inventory
- Order processing with line items and status tracking
- Full CRUD through a repository pattern
- Database migrations with Alembic
- Both sync and async database operations
Data Model
┌─────────────────────────────────────────────────────────────────────────────┐
│ │
│ ┌───────────────┐ ┌────────────────┐ ┌───────────────┐ │
│ │ Category │ 1 N │ Product │ │ User │ │
│ │───────────────│◄──────│────────────────│ │───────────────│ │
│ │ id (PK) │ │ id (PK) │ │ id (PK) │ │
│ │ name │ │ name │ │ email (unique)│ │
│ │ description │ │ description │ │ name │ │
│ │ created_at │ │ price │ │ hashed_pwd │ │
│ │ │ │ stock_quantity │ │ is_active │ │
│ │ │ │ category_id(FK)│ │ created_at │ │
│ │ │ │ created_at │ │ │ │
│ └───────────────┘ └───────┬────────┘ └───────┬───────┘ │
│ │ │ │
│ │ N │ 1 │
│ │ │ │
│ ▼ ▼ │
│ ┌────────────────┐ ┌───────────────┐ │
│ │ OrderItem │ N 1 │ Order │ │
│ │────────────────│──────►│───────────────│ │
│ │ id (PK) │ │ id (PK) │ │
│ │ order_id (FK) │ │ user_id (FK) │ │
│ │ product_id(FK) │ │ status │ │
│ │ quantity │ │ total_amount │ │
│ │ unit_price │ │ created_at │ │
│ │ │ │ updated_at │ │
│ └────────────────┘ └───────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
API Endpoints
Categories:
POST /categories Create category
GET /categories List categories
GET /categories/{id} Get category with products
PATCH /categories/{id} Update category
DELETE /categories/{id} Delete category
Products:
POST /products Create product
GET /products List products (with filters)
GET /products/{id} Get product details
PATCH /products/{id} Update product
DELETE /products/{id} Delete product
POST /products/{id}/stock Adjust stock
Users:
POST /users Register user
GET /users/{id} Get user profile
PATCH /users/{id} Update user
DELETE /users/{id} Deactivate user
Orders:
POST /orders Create order
GET /orders List orders (with filters)
GET /orders/{id} Get order with items
PATCH /orders/{id}/status Update order status
DELETE /orders/{id} Cancel order
Solution Architecture
Project Structure
ecommerce-sqlmodel/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI application
│ ├── config.py # Settings with pydantic-settings
│ │
│ ├── models/ # SQLModel models
│ │ ├── __init__.py
│ │ ├── base.py # Base classes and common fields
│ │ ├── user.py # User models (table + schemas)
│ │ ├── category.py # Category models
│ │ ├── product.py # Product models
│ │ └── order.py # Order and OrderItem models
│ │
│ ├── repositories/ # Data access layer
│ │ ├── __init__.py
│ │ ├── base.py # Generic repository base
│ │ ├── user_repository.py
│ │ ├── category_repository.py
│ │ ├── product_repository.py
│ │ └── order_repository.py
│ │
│ ├── routes/ # API routes
│ │ ├── __init__.py
│ │ ├── users.py
│ │ ├── categories.py
│ │ ├── products.py
│ │ └── orders.py
│ │
│ ├── services/ # Business logic
│ │ ├── __init__.py
│ │ ├── user_service.py
│ │ ├── product_service.py
│ │ └── order_service.py
│ │
│ └── db/ # Database configuration
│ ├── __init__.py
│ ├── engine.py # Engine and session
│ └── init_db.py # Database initialization
│
├── alembic/ # Migrations
│ ├── versions/
│ ├── env.py
│ └── alembic.ini
│
├── tests/
│ ├── __init__.py
│ ├── conftest.py # Fixtures
│ ├── test_models.py
│ ├── test_repositories.py
│ └── test_api.py
│
├── pyproject.toml
└── README.md
Component Diagram
┌─────────────────────────────────────────────────────────────────────────────┐
│ main.py │
│ - FastAPI app │
│ - Exception handlers │
│ - Lifespan events (DB init) │
└─────────────────────────────────────┬───────────────────────────────────────┘
│
┌───────────────────────┼───────────────────────┐
▼ ▼ ▼
┌───────────────────────┐ ┌───────────────────────┐ ┌───────────────────────┐
│ routes/users.py │ │ routes/products.py │ │ routes/orders.py │
│ - Input validation │ │ - Input validation │ │ - Input validation │
│ - HTTP handling │ │ - Query params │ │ - Status transitions │
└───────────┬───────────┘ └───────────┬───────────┘ └───────────┬───────────┘
│ │ │
▼ ▼ ▼
┌───────────────────────┐ ┌───────────────────────┐ ┌───────────────────────┐
│ services/user_svc.py │ │ services/product_svc │ │ services/order_svc │
│ - Business rules │ │ - Stock management │ │ - Order creation │
│ - Password hashing │ │ - Price calculation │ │ - Total calculation │
└───────────┬───────────┘ └───────────┬───────────┘ └───────────┬───────────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ repositories/ │
│ - UserRepository - ProductRepository - OrderRepository │
│ - Generic CRUD - Stock queries - Order with items │
│ - SQL abstraction - Category filters - Status updates │
└─────────────────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ db/engine.py │
│ - SQLModel engine - Session factory - Async support │
└─────────────────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────┐
│ PostgreSQL │
│ Database │
└─────────────────┘
Data Flow for Order Creation
POST /orders
{
"user_id": 1,
"items": [
{"product_id": 101, "quantity": 2},
{"product_id": 102, "quantity": 1}
]
}
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ Route: create_order(order_create: OrderCreate) │
│ - Validate input with Pydantic │
│ - Call OrderService │
└─────────────────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ OrderService.create_order() │
│ 1. Verify user exists │
│ 2. Verify products exist and have stock │
│ 3. Calculate total from product prices × quantities │
│ 4. Create Order and OrderItems in transaction │
│ 5. Deduct stock from products │
└─────────────────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ OrderRepository.create_with_items(order, items) │
│ session.begin() │
│ session.add(order) │
│ session.add_all(order_items) │
│ session.commit() │
│ session.refresh(order) │
└─────────────────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ Response: OrderRead │
│ { │
│ "id": 1, │
│ "user_id": 1, │
│ "status": "pending", │
│ "total_amount": 149.97, │
│ "items": [ │
│ {"product_id": 101, "quantity": 2, "unit_price": 49.99}, │
│ {"product_id": 102, "quantity": 1, "unit_price": 49.99} │
│ ] │
│ } │
└─────────────────────────────────────────────────────────────────────────────┘
Phased Implementation Guide
Phase 1: Project Setup and Base Models (1-2 hours)
Goal: Set up the project structure with database connection.
Step 1: Create project and install dependencies:
mkdir ecommerce-sqlmodel && cd ecommerce-sqlmodel
python -m venv venv
source venv/bin/activate
pip install fastapi uvicorn sqlmodel alembic asyncpg psycopg2-binary pydantic-settings python-dotenv
Step 2: Create database configuration:
# app/config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str = "postgresql://user:pass@localhost/ecommerce"
database_echo: bool = False
class Config:
env_file = ".env"
settings = Settings()
Step 3: Create database engine:
# app/db/engine.py
from sqlmodel import SQLModel, create_engine, Session
from app.config import settings
engine = create_engine(settings.database_url, echo=settings.database_echo)
def get_session():
with Session(engine) as session:
yield session
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
Step 4: Create base model:
# app/models/base.py
from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional
class TimestampMixin(SQLModel):
"""Mixin for created_at and updated_at fields."""
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: Optional[datetime] = Field(default=None)
Checkpoint: Can connect to database and create tables.
Phase 2: User Models and Repository (2 hours)
Goal: Implement complete user management with the repository pattern.
Step 1: Create user models:
# app/models/user.py
from sqlmodel import SQLModel, Field, Relationship
from pydantic import EmailStr, ConfigDict
from typing import Optional, List, TYPE_CHECKING
from datetime import datetime
if TYPE_CHECKING:
from .order import Order
# Base fields shared across models
class UserBase(SQLModel):
email: EmailStr = Field(unique=True, index=True)
name: str = Field(min_length=1, max_length=100)
is_active: bool = Field(default=True)
# Database table
class User(UserBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
hashed_password: str
created_at: datetime = Field(default_factory=datetime.utcnow)
# Relationships
orders: List["Order"] = Relationship(back_populates="user")
# API Input: Create
class UserCreate(UserBase):
password: str = Field(min_length=8, max_length=100)
# API Input: Update
class UserUpdate(SQLModel):
email: Optional[EmailStr] = None
name: Optional[str] = Field(None, min_length=1, max_length=100)
password: Optional[str] = Field(None, min_length=8, max_length=100)
is_active: Optional[bool] = None
# API Output
class UserRead(UserBase):
id: int
created_at: datetime
model_config = ConfigDict(from_attributes=True)
Step 2: Create user repository:
# app/repositories/user_repository.py
from sqlmodel import Session, select
from typing import Optional, List
from app.models.user import User, UserCreate, UserUpdate
import hashlib
def hash_password(password: str) -> str:
"""Simple hash for demo. Use bcrypt in production!"""
return hashlib.sha256(password.encode()).hexdigest()
class UserRepository:
def __init__(self, session: Session):
self.session = session
def create(self, user_create: UserCreate) -> User:
user = User(
email=user_create.email,
name=user_create.name,
hashed_password=hash_password(user_create.password)
)
self.session.add(user)
self.session.commit()
self.session.refresh(user)
return user
def get_by_id(self, user_id: int) -> Optional[User]:
return self.session.get(User, user_id)
def get_by_email(self, email: str) -> Optional[User]:
statement = select(User).where(User.email == email)
return self.session.exec(statement).first()
def list(self, skip: int = 0, limit: int = 100) -> List[User]:
statement = select(User).offset(skip).limit(limit)
return self.session.exec(statement).all()
def update(self, user_id: int, user_update: UserUpdate) -> Optional[User]:
user = self.get_by_id(user_id)
if not user:
return None
update_data = user_update.model_dump(exclude_unset=True)
if "password" in update_data:
update_data["hashed_password"] = hash_password(update_data.pop("password"))
for key, value in update_data.items():
setattr(user, key, value)
self.session.add(user)
self.session.commit()
self.session.refresh(user)
return user
def delete(self, user_id: int) -> bool:
user = self.get_by_id(user_id)
if not user:
return False
self.session.delete(user)
self.session.commit()
return True
Checkpoint: Can create, read, update, delete users through repository.
Phase 3: Product and Category Models (2 hours)
Goal: Implement product catalog with category relationships.
Step 1: Create category model:
# app/models/category.py
from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List, TYPE_CHECKING
from datetime import datetime
from pydantic import ConfigDict
if TYPE_CHECKING:
from .product import Product
class CategoryBase(SQLModel):
name: str = Field(min_length=1, max_length=100)
description: Optional[str] = Field(None, max_length=500)
class Category(CategoryBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
created_at: datetime = Field(default_factory=datetime.utcnow)
products: List["Product"] = Relationship(back_populates="category")
class CategoryCreate(CategoryBase):
pass
class CategoryRead(CategoryBase):
id: int
created_at: datetime
product_count: int = 0
model_config = ConfigDict(from_attributes=True)
class CategoryReadWithProducts(CategoryRead):
products: List["ProductRead"] = []
Step 2: Create product model:
# app/models/product.py
from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, TYPE_CHECKING
from datetime import datetime
from decimal import Decimal
from pydantic import ConfigDict
if TYPE_CHECKING:
from .category import Category
class ProductBase(SQLModel):
name: str = Field(min_length=1, max_length=200)
description: Optional[str] = Field(None, max_length=2000)
price: Decimal = Field(ge=0, decimal_places=2)
stock_quantity: int = Field(ge=0, default=0)
class Product(ProductBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
category_id: Optional[int] = Field(default=None, foreign_key="category.id")
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: Optional[datetime] = None
category: Optional["Category"] = Relationship(back_populates="products")
class ProductCreate(ProductBase):
category_id: Optional[int] = None
class ProductUpdate(SQLModel):
name: Optional[str] = Field(None, min_length=1, max_length=200)
description: Optional[str] = None
price: Optional[Decimal] = Field(None, ge=0)
stock_quantity: Optional[int] = Field(None, ge=0)
category_id: Optional[int] = None
class ProductRead(ProductBase):
id: int
category_id: Optional[int]
created_at: datetime
model_config = ConfigDict(from_attributes=True)
class ProductReadWithCategory(ProductRead):
category: Optional["CategoryRead"] = None
Checkpoint: Can manage products and categories with relationships.
Phase 4: Order Models with Transactions (2-3 hours)
Goal: Implement order processing with proper transaction handling.
Step 1: Create order models:
# app/models/order.py
from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List, TYPE_CHECKING
from datetime import datetime
from decimal import Decimal
from enum import Enum
from pydantic import ConfigDict
if TYPE_CHECKING:
from .user import User
from .product import Product
class OrderStatus(str, Enum):
pending = "pending"
confirmed = "confirmed"
processing = "processing"
shipped = "shipped"
delivered = "delivered"
cancelled = "cancelled"
class OrderItemBase(SQLModel):
quantity: int = Field(ge=1)
unit_price: Decimal = Field(ge=0, decimal_places=2)
class OrderItem(OrderItemBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
order_id: int = Field(foreign_key="order.id")
product_id: int = Field(foreign_key="product.id")
order: Optional["Order"] = Relationship(back_populates="items")
product: Optional["Product"] = Relationship()
class OrderBase(SQLModel):
status: OrderStatus = Field(default=OrderStatus.pending)
total_amount: Decimal = Field(ge=0, decimal_places=2, default=0)
class Order(OrderBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
user_id: int = Field(foreign_key="user.id")
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: Optional[datetime] = None
user: Optional["User"] = Relationship(back_populates="orders")
items: List["OrderItem"] = Relationship(back_populates="order")
# API Input
class OrderItemCreate(SQLModel):
product_id: int
quantity: int = Field(ge=1)
class OrderCreate(SQLModel):
user_id: int
items: List[OrderItemCreate]
# API Output
class OrderItemRead(OrderItemBase):
id: int
product_id: int
model_config = ConfigDict(from_attributes=True)
class OrderRead(OrderBase):
id: int
user_id: int
created_at: datetime
items: List[OrderItemRead] = []
model_config = ConfigDict(from_attributes=True)
Step 2: Create order service with transaction:
# app/services/order_service.py
from sqlmodel import Session, select
from typing import Optional
from decimal import Decimal
from app.models.order import Order, OrderItem, OrderCreate, OrderStatus
from app.models.product import Product
from app.models.user import User
class InsufficientStockError(Exception):
def __init__(self, product_id: int, requested: int, available: int):
self.product_id = product_id
self.requested = requested
self.available = available
super().__init__(f"Product {product_id}: requested {requested}, available {available}")
class OrderService:
def __init__(self, session: Session):
self.session = session
def create_order(self, order_create: OrderCreate) -> Order:
# Verify user exists
user = self.session.get(User, order_create.user_id)
if not user:
raise ValueError(f"User {order_create.user_id} not found")
# Verify products and calculate total
total = Decimal("0.00")
order_items = []
for item in order_create.items:
product = self.session.get(Product, item.product_id)
if not product:
raise ValueError(f"Product {item.product_id} not found")
if product.stock_quantity < item.quantity:
raise InsufficientStockError(
item.product_id,
item.quantity,
product.stock_quantity
)
item_total = product.price * item.quantity
total += item_total
order_items.append({
"product": product,
"quantity": item.quantity,
"unit_price": product.price
})
# Create order
order = Order(
user_id=order_create.user_id,
total_amount=total,
status=OrderStatus.pending
)
self.session.add(order)
self.session.flush() # Get order.id without committing
# Create order items and deduct stock
for item_data in order_items:
order_item = OrderItem(
order_id=order.id,
product_id=item_data["product"].id,
quantity=item_data["quantity"],
unit_price=item_data["unit_price"]
)
self.session.add(order_item)
# Deduct stock
item_data["product"].stock_quantity -= item_data["quantity"]
self.session.add(item_data["product"])
self.session.commit()
self.session.refresh(order)
return order
def update_status(self, order_id: int, new_status: OrderStatus) -> Optional[Order]:
order = self.session.get(Order, order_id)
if not order:
return None
# Status transition validation
valid_transitions = {
OrderStatus.pending: [OrderStatus.confirmed, OrderStatus.cancelled],
OrderStatus.confirmed: [OrderStatus.processing, OrderStatus.cancelled],
OrderStatus.processing: [OrderStatus.shipped, OrderStatus.cancelled],
OrderStatus.shipped: [OrderStatus.delivered],
OrderStatus.delivered: [],
OrderStatus.cancelled: []
}
if new_status not in valid_transitions.get(order.status, []):
raise ValueError(f"Cannot transition from {order.status} to {new_status}")
order.status = new_status
self.session.add(order)
self.session.commit()
self.session.refresh(order)
return order
Checkpoint: Can create orders with stock validation and transactions.
Phase 5: Alembic Migrations (1-2 hours)
Goal: Set up and use Alembic for schema migrations.
Step 1: Initialize Alembic:
alembic init alembic
Step 2: Configure Alembic to use SQLModel:
# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
from sqlmodel import SQLModel
# Import all models so they're registered with SQLModel.metadata
from app.models.user import User
from app.models.category import Category
from app.models.product import Product
from app.models.order import Order, OrderItem
# This is the Alembic Config object
config = context.config
# Interpret the config file for Python logging
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# Use SQLModel metadata
target_metadata = SQLModel.metadata
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode."""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode."""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
Step 3: Generate initial migration:
alembic revision --autogenerate -m "Initial schema"
alembic upgrade head
Checkpoint: Can generate and apply database migrations.
Phase 6: FastAPI Integration and Testing (2-3 hours)
Goal: Wire everything together with FastAPI routes and tests.
Step 1: Create the main application:
# app/main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session
from contextlib import asynccontextmanager
from app.db.engine import create_db_and_tables, get_session
from app.routes import users, products, categories, orders
@asynccontextmanager
async def lifespan(app: FastAPI):
create_db_and_tables()
yield
app = FastAPI(
title="E-Commerce API",
description="SQLModel-powered e-commerce backend",
version="1.0.0",
lifespan=lifespan
)
app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(categories.router, prefix="/categories", tags=["categories"])
app.include_router(products.router, prefix="/products", tags=["products"])
app.include_router(orders.router, prefix="/orders", tags=["orders"])
@app.get("/health")
async def health():
return {"status": "healthy"}
Step 2: Create routes:
# app/routes/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session
from typing import List
from app.db.engine import get_session
from app.models.user import UserCreate, UserRead, UserUpdate
from app.repositories.user_repository import UserRepository
router = APIRouter()
@router.post("", response_model=UserRead, status_code=201)
def create_user(
user: UserCreate,
session: Session = Depends(get_session)
):
repo = UserRepository(session)
if repo.get_by_email(user.email):
raise HTTPException(409, "Email already registered")
return repo.create(user)
@router.get("", response_model=List[UserRead])
def list_users(
skip: int = 0,
limit: int = 100,
session: Session = Depends(get_session)
):
repo = UserRepository(session)
return repo.list(skip, limit)
@router.get("/{user_id}", response_model=UserRead)
def get_user(user_id: int, session: Session = Depends(get_session)):
repo = UserRepository(session)
user = repo.get_by_id(user_id)
if not user:
raise HTTPException(404, "User not found")
return user
@router.patch("/{user_id}", response_model=UserRead)
def update_user(
user_id: int,
user_update: UserUpdate,
session: Session = Depends(get_session)
):
repo = UserRepository(session)
user = repo.update(user_id, user_update)
if not user:
raise HTTPException(404, "User not found")
return user
@router.delete("/{user_id}", status_code=204)
def delete_user(user_id: int, session: Session = Depends(get_session)):
repo = UserRepository(session)
if not repo.delete(user_id):
raise HTTPException(404, "User not found")
Checkpoint: Full API working with all CRUD operations.
Testing Strategy
Unit Tests for Models
# tests/test_models.py
import pytest
from pydantic import ValidationError
from app.models.user import UserCreate, UserUpdate, UserRead
from app.models.order import OrderStatus
def test_user_create_valid():
user = UserCreate(
email="test@example.com",
name="Test User",
password="securepassword123"
)
assert user.email == "test@example.com"
assert user.name == "Test User"
def test_user_create_invalid_email():
with pytest.raises(ValidationError) as exc:
UserCreate(
email="not-an-email",
name="Test",
password="password123"
)
errors = exc.value.errors()
assert any(e["loc"] == ("email",) for e in errors)
def test_user_create_short_password():
with pytest.raises(ValidationError) as exc:
UserCreate(
email="test@example.com",
name="Test",
password="short"
)
errors = exc.value.errors()
assert any(e["loc"] == ("password",) for e in errors)
def test_user_update_partial():
update = UserUpdate(name="New Name")
assert update.name == "New Name"
assert update.email is None
assert update.password is None
def test_order_status_enum():
assert OrderStatus.pending.value == "pending"
assert OrderStatus.cancelled.value == "cancelled"
Repository Tests
# tests/test_repositories.py
import pytest
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool
from app.models.user import UserCreate, UserUpdate
from app.repositories.user_repository import UserRepository
@pytest.fixture
def session():
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
yield session
def test_create_user(session):
repo = UserRepository(session)
user = repo.create(UserCreate(
email="test@example.com",
name="Test User",
password="password123"
))
assert user.id is not None
assert user.email == "test@example.com"
assert user.hashed_password != "password123"
def test_get_user_by_email(session):
repo = UserRepository(session)
repo.create(UserCreate(
email="find@example.com",
name="Find Me",
password="password123"
))
user = repo.get_by_email("find@example.com")
assert user is not None
assert user.name == "Find Me"
def test_update_user(session):
repo = UserRepository(session)
user = repo.create(UserCreate(
email="update@example.com",
name="Original",
password="password123"
))
updated = repo.update(user.id, UserUpdate(name="Updated"))
assert updated.name == "Updated"
assert updated.email == "update@example.com"
def test_delete_user(session):
repo = UserRepository(session)
user = repo.create(UserCreate(
email="delete@example.com",
name="Delete Me",
password="password123"
))
assert repo.delete(user.id) is True
assert repo.get_by_id(user.id) is None
Integration Tests
# tests/test_api.py
import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool
from app.main import app
from app.db.engine import get_session
@pytest.fixture
def client():
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool
)
SQLModel.metadata.create_all(engine)
def get_test_session():
with Session(engine) as session:
yield session
app.dependency_overrides[get_session] = get_test_session
with TestClient(app) as client:
yield client
app.dependency_overrides.clear()
def test_create_user_api(client):
response = client.post("/users", json={
"email": "api@example.com",
"name": "API User",
"password": "password123"
})
assert response.status_code == 201
data = response.json()
assert data["email"] == "api@example.com"
assert "password" not in data
assert "id" in data
def test_create_user_duplicate_email(client):
client.post("/users", json={
"email": "dupe@example.com",
"name": "First",
"password": "password123"
})
response = client.post("/users", json={
"email": "dupe@example.com",
"name": "Second",
"password": "password123"
})
assert response.status_code == 409
def test_create_order_insufficient_stock(client):
# Create user
user_resp = client.post("/users", json={
"email": "buyer@example.com",
"name": "Buyer",
"password": "password123"
})
user_id = user_resp.json()["id"]
# Create product with limited stock
cat_resp = client.post("/categories", json={"name": "Test"})
prod_resp = client.post("/products", json={
"name": "Limited Product",
"price": 10.00,
"stock_quantity": 5,
"category_id": cat_resp.json()["id"]
})
product_id = prod_resp.json()["id"]
# Try to order more than available
response = client.post("/orders", json={
"user_id": user_id,
"items": [{"product_id": product_id, "quantity": 10}]
})
assert response.status_code == 400
assert "insufficient" in response.json()["detail"].lower()
Common Pitfalls and Debugging
Pitfall 1: Forgetting table=True
Problem: Model doesn’t create database table.
# Wrong: Missing table=True
class User(SQLModel):
id: int = Field(primary_key=True)
name: str
# Correct
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
Symptom: sqlalchemy.exc.NoSuchTableError: user
Pitfall 2: Optional Primary Key
Problem: Primary key must be Optional for auto-generation.
# Wrong: Can't insert without id
class User(SQLModel, table=True):
id: int = Field(primary_key=True) # Required!
# Correct: Optional with default None
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
Pitfall 3: Circular Import with Relationships
Problem: Models reference each other causing import errors.
# Wrong: Direct import causes circular dependency
from app.models.order import Order
class User(SQLModel, table=True):
orders: List[Order] # Error!
# Correct: Use TYPE_CHECKING and string annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from app.models.order import Order
class User(SQLModel, table=True):
orders: List["Order"] = Relationship(back_populates="user")
Pitfall 4: Missing from_attributes
Problem: Can’t return ORM objects as response models.
# Wrong: Missing from_attributes
class UserRead(SQLModel):
id: int
name: str
@app.get("/users/{id}", response_model=UserRead)
def get_user(id: int, session: Session = Depends(get_session)):
return session.get(User, id) # Error!
# Correct: Add from_attributes
class UserRead(SQLModel):
id: int
name: str
model_config = ConfigDict(from_attributes=True)
Pitfall 5: N+1 Query Problem
Problem: Loading relationships causes too many queries.
# Wrong: N+1 queries when accessing order.items for each order
orders = session.exec(select(Order)).all()
for order in orders:
print(order.items) # Each access triggers a query!
# Correct: Use selectinload or joinedload
from sqlalchemy.orm import selectinload
statement = select(Order).options(selectinload(Order.items))
orders = session.exec(statement).all()
# All items loaded in 2 queries total
Pitfall 6: Session Not Refreshed
Problem: Object doesn’t have updated values after commit.
# Wrong: user.id is still None after commit
user = User(name="Test")
session.add(user)
session.commit()
print(user.id) # None!
# Correct: Refresh to get database-generated values
session.add(user)
session.commit()
session.refresh(user)
print(user.id) # 1
Pitfall 7: Alembic Not Detecting Changes
Problem: Autogenerate doesn’t see model changes.
Solution: Ensure all models are imported in alembic/env.py:
# alembic/env.py
# Import ALL models here
from app.models.user import User
from app.models.order import Order, OrderItem
from app.models.product import Product
from app.models.category import Category
target_metadata = SQLModel.metadata
Extensions and Challenges
Extension 1: Async Database Operations
Convert to async for better performance:
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
async_session = async_sessionmaker(engine, class_=AsyncSession)
async def get_async_session() -> AsyncSession:
async with async_session() as session:
yield session
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
session: AsyncSession = Depends(get_async_session)
):
result = await session.exec(select(User).where(User.id == user_id))
return result.first()
Extension 2: Soft Deletes
Implement soft delete pattern:
class SoftDeleteMixin(SQLModel):
deleted_at: Optional[datetime] = None
is_deleted: bool = Field(default=False)
class User(UserBase, SoftDeleteMixin, table=True):
...
# Repository method
def soft_delete(self, user_id: int) -> bool:
user = self.get_by_id(user_id)
if not user:
return False
user.is_deleted = True
user.deleted_at = datetime.utcnow()
self.session.add(user)
self.session.commit()
return True
# Query excludes soft-deleted
def list_active(self):
statement = select(User).where(User.is_deleted == False)
return self.session.exec(statement).all()
Extension 3: Query Builder Pattern
Build complex queries fluently:
class ProductQueryBuilder:
def __init__(self, session: Session):
self.session = session
self.statement = select(Product)
def in_category(self, category_id: int) -> "ProductQueryBuilder":
self.statement = self.statement.where(
Product.category_id == category_id
)
return self
def in_stock(self) -> "ProductQueryBuilder":
self.statement = self.statement.where(
Product.stock_quantity > 0
)
return self
def price_range(self, min_price: Decimal, max_price: Decimal):
self.statement = self.statement.where(
Product.price.between(min_price, max_price)
)
return self
def execute(self) -> List[Product]:
return self.session.exec(self.statement).all()
# Usage
products = (
ProductQueryBuilder(session)
.in_category(1)
.in_stock()
.price_range(10, 100)
.execute()
)
Extension 4: Event Sourcing for Orders
Track all order state changes:
class OrderEvent(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
order_id: int = Field(foreign_key="order.id")
event_type: str # "created", "confirmed", "shipped", etc.
old_status: Optional[str]
new_status: str
metadata: dict = Field(default={}, sa_column=Column(JSON))
created_at: datetime = Field(default_factory=datetime.utcnow)
created_by: Optional[int] = None
def update_order_status(order_id: int, new_status: OrderStatus, user_id: int):
order = session.get(Order, order_id)
old_status = order.status
# Create event
event = OrderEvent(
order_id=order_id,
event_type="status_change",
old_status=old_status,
new_status=new_status,
created_by=user_id
)
session.add(event)
# Update order
order.status = new_status
session.add(order)
session.commit()
Extension 5: Full-Text Search
Add search capability with PostgreSQL:
from sqlalchemy import func
class ProductRepository:
def search(self, query: str) -> List[Product]:
# PostgreSQL full-text search
statement = select(Product).where(
func.to_tsvector('english', Product.name + ' ' + Product.description)
.match(query)
)
return self.session.exec(statement).all()
Real-World Connections
Where This Pattern Appears
- FastAPI + SQLModel - Official recommended stack
- Django ORM - Similar model-first approach
- ActiveRecord (Rails) - Models as tables pattern
- Entity Framework (.NET) - Code-first migrations
Industry Examples
- Stripe - Repository pattern for payment models
- GitHub - Event sourcing for repository events
- Shopify - Order and inventory management
- Netflix - Async database operations for streaming
Production Considerations
- Connection Pooling - Use SQLAlchemy pool settings
- Read Replicas - Route reads to replicas
- Migrations in CI/CD - Run migrations before deployment
- Backup Strategy - Regular database backups
- Monitoring - Query performance and connection metrics
Self-Assessment Checklist
Core Understanding
- Can I explain the difference between table=True and table=False?
- Can I describe how SQLModel combines Pydantic and SQLAlchemy?
- Can I explain why we separate input, output, and database models?
- Can I describe the repository pattern and its benefits?
Implementation Skills
- Can I create SQLModel tables with relationships?
- Can I implement a repository with full CRUD operations?
- Can I handle transactions for multi-step operations?
- Can I set up and run Alembic migrations?
Advanced Skills
- Can I implement async database operations?
- Can I prevent N+1 query problems?
- Can I handle circular imports between models?
- Can I write comprehensive tests for repositories?
Mastery Indicators
- Order creation properly validates and deducts stock
- Migrations generate correctly from model changes
- Tests cover both success and error cases
- API returns properly serialized responses
Resources
Documentation
Books
- “Architecture Patterns with Python” by Harry Percival & Bob Gregory
- “Cosmic Python” (free online) - Domain-driven design with Python
Related Projects
Videos
- Sebastián Ramírez’s SQLModel talks
- ArjanCodes - SQLModel tutorials