Python SQLModel

      Ankush More

      Table of Contents

      1. Introduction to SQLModel
      2. Setting Up Your Environment
      3. Basic Concepts
      4. Creating Your First Model
      5. Database Operations
      6. Relationships
      7. Advanced Queries
      8. FastAPI Integration
      9. Testing
      10. Performance Optimization
      11. Production Deployment
      12. Best Practices

      1. Introduction to SQLModel

      SQLModel is a library created by Sebastian Ramirez (the creator of FastAPI) that combines the power of SQLAlchemy and Pydantic. It allows you to define database models using Python type hints that work seamlessly with both database operations and API serialization.

      Why SQLModel?

      graph TD
    A[Traditional Approach] --> B[SQLAlchemy Models]
    A --> C[Pydantic Models]
    A --> D[Duplicate Code]

    E[SQLModel Approach] --> F[Single Model Definition]
    F --> G[Database Operations]
    F --> H[API Serialization]
    F --> I[Type Safety]

      Key Benefits

      • Single Source of Truth: One model for database and API
      • Type Safety: Full TypeScript-like type checking
      • FastAPI Integration: Seamless API development
      • SQLAlchemy Power: All SQLAlchemy features available

      2. Setting Up Your Environment

      Installation

      # Basic installation
pip install sqlmodel

# With database drivers
pip install sqlmodel[all]

# For PostgreSQL
pip install sqlmodel psycopg2-binary

# For MySQL
pip install sqlmodel pymysql

# For development
pip install sqlmodel[all] pytest pytest-asyncio
      Bash

      Project Structure

      graph TD
    A[Project Root] --> B[app/]
    B --> C[models.py]
    B --> D[database.py]
    B --> E[main.py]
    A --> F[tests/]
    F --> G[test_models.py]
    A --> H[requirements.txt]
    A --> I[.env]

      Basic Setup

      # database.py
from sqlmodel import SQLModel, create_engine, Session
from typing import Optional

# Database URL - can be SQLite, PostgreSQL, MySQL, etc.
DATABASE_URL = "sqlite:///./test.db"

# Create engine
engine = create_engine(DATABASE_URL, echo=True)

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

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

      3. Basic Concepts

      Understanding SQLModel Architecture

      classDiagram
    class SQLModel {
        +table: bool
        +registry: registry
        +metadata: MetaData
        +__init__()
        +model_validate()
        +model_dump()
    }

    class Pydantic {
        +BaseModel
        +Field validation
        +JSON serialization
    }

    class SQLAlchemy {
        +DeclarativeBase
        +Table mapping
        +Query interface
    }

    SQLModel --|> Pydantic
    SQLModel --|> SQLAlchemy

      Core Components

      1. Models: Define your data structure
      2. Engine: Database connection
      3. Session: Database transaction handler
      4. Fields: Column definitions with validation

      4. Creating Your First Model

      Simple Model Example

      # models.py
from sqlmodel import SQLModel, Field
from typing import Optional
from datetime import datetime

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True, max_length=50)
    age: Optional[int] = Field(default=None, ge=0, le=150)
    secret_name: str
    created_at: datetime = Field(default_factory=datetime.utcnow)

    class Config:
        # Example configuration
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }
      Python

      Model with Validation

      from sqlmodel import SQLModel, Field
from typing import Optional
from enum import Enum

class PowerLevel(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    EXTREME = "extreme"

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(min_length=1, max_length=50, index=True)
    age: Optional[int] = Field(default=None, ge=0, le=200)
    power_level: PowerLevel = Field(default=PowerLevel.LOW)
    is_active: bool = Field(default=True)

    # Custom validator
    @validator('name')
    def validate_name(cls, v):
        if not v.strip():
            raise ValueError('Name cannot be empty')
        return v.title()
      Python

      Model Inheritance

      class BaseModel(SQLModel):
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: Optional[datetime] = Field(default=None)

class Hero(BaseModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str

class Team(BaseModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(unique=True)
    headquarters: str
      Python

      5. Database Operations

      CRUD Operations Flow

      sequenceDiagram
    participant App
    participant Session
    participant Database

    App->>Session: Create session
    App->>Session: add(model)
    Session->>Database: INSERT
    App->>Session: commit()
    Session->>Database: COMMIT
    App->>Session: close()

      Create Operations

      # create_hero.py
from sqlmodel import Session
from models import Hero
from database import engine

def create_hero(hero_data: dict) -> Hero:
    with Session(engine) as session:
        hero = Hero(**hero_data)
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero

# Usage
new_hero = create_hero({
    "name": "Spider-Man",
    "secret_name": "Peter Parker",
    "age": 25
})
      Python

      Read Operations

      def get_hero_by_id(hero_id: int) -> Optional[Hero]:
    with Session(engine) as session:
        return session.get(Hero, hero_id)

def get_heroes_by_name(name: str) -> List[Hero]:
    with Session(engine) as session:
        statement = select(Hero).where(Hero.name.contains(name))
        return session.exec(statement).all()

def get_all_heroes(skip: int = 0, limit: int = 100) -> List[Hero]:
    with Session(engine) as session:
        statement = select(Hero).offset(skip).limit(limit)
        return session.exec(statement).all()
      Python

      Update Operations

      def update_hero(hero_id: int, hero_update: dict) -> Optional[Hero]:
    with Session(engine) as session:
        hero = session.get(Hero, hero_id)
        if not hero:
            return None

        for key, value in hero_update.items():
            setattr(hero, key, value)

        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero
      Python

      Delete Operations

      def delete_hero(hero_id: int) -> bool:
    with Session(engine) as session:
        hero = session.get(Hero, hero_id)
        if not hero:
            return False

        session.delete(hero)
        session.commit()
        return True
      Python

      6. Relationships

      Relationship Types Overview

      erDiagram
    HERO ||--o{ TEAM_HERO_LINK : "belongs to"
    TEAM ||--o{ TEAM_HERO_LINK : "has"
    HERO ||--o{ POWER : "has"
    HERO ||--|| IDENTITY : "has"

    HERO {
        int id PK
        string name
        string secret_name
    }

    TEAM {
        int id PK
        string name
        string headquarters
    }

    POWER {
        int id PK
        string name
        int hero_id FK
    }

    IDENTITY {
        int id PK
        string real_name
        int hero_id FK
    }

      One-to-Many Relationship

      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 = Field(unique=True)
    headquarters: str

    # Relationship
    heroes: List["Hero"] = Relationship(back_populates="team")

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

    # Relationship
    team: Optional[Team] = Relationship(back_populates="heroes")
      Python

      Many-to-Many Relationship

      class HeroTeamLink(SQLModel, table=True):
    team_id: Optional[int] = Field(
        default=None, foreign_key="team.id", primary_key=True
    )
    hero_id: Optional[int] = Field(
        default=None, foreign_key="hero.id", primary_key=True
    )

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    headquarters: str

    heroes: List["Hero"] = Relationship(
        back_populates="teams", link_model=HeroTeamLink
    )

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str

    teams: List[Team] = Relationship(
        back_populates="heroes", link_model=HeroTeamLink
    )
      Python

      Working with Relationships

      def create_team_with_heroes():
    with Session(engine) as session:
        # Create team
        team = Team(name="Avengers", headquarters="Avengers Tower")
        session.add(team)
        session.commit()
        session.refresh(team)

        # Create heroes and add to team
        heroes_data = [
            {"name": "Iron Man", "secret_name": "Tony Stark"},
            {"name": "Captain America", "secret_name": "Steve Rogers"}
        ]

        for hero_data in heroes_data:
            hero = Hero(**hero_data, team_id=team.id)
            session.add(hero)

        session.commit()
        return team

def get_team_with_heroes(team_id: int):
    with Session(engine) as session:
        statement = select(Team).where(Team.id == team_id)
        team = session.exec(statement).first()

        # Access related heroes (lazy loading)
        if team:
            print(f"Team: {team.name}")
            for hero in team.heroes:
                print(f"  Hero: {hero.name}")

        return team
      Python

      7. Advanced Queries

      Query Building Pattern

      graph LR
    A["select()"] --> B["where()"]
    B --> C["join()"]
    C --> D["order_by()"]
    D --> E["limit()"]
    E --> F["exec()"]

      Complex Queries

      from sqlmodel import select, and_, or_, func, desc

def advanced_hero_search(
    name_contains: Optional[str] = None,
    min_age: Optional[int] = None,
    team_name: Optional[str] = None,
    power_level: Optional[PowerLevel] = None,
    skip: int = 0,
    limit: int = 100
):
    with Session(engine) as session:
        statement = select(Hero)

        # Build WHERE conditions
        conditions = []

        if name_contains:
            conditions.append(Hero.name.contains(name_contains))

        if min_age:
            conditions.append(Hero.age >= min_age)

        if power_level:
            conditions.append(Hero.power_level == power_level)

        if team_name:
            statement = statement.join(Team)
            conditions.append(Team.name == team_name)

        if conditions:
            statement = statement.where(and_(*conditions))

        # Add ordering and pagination
        statement = statement.order_by(desc(Hero.created_at))
        statement = statement.offset(skip).limit(limit)

        return session.exec(statement).all()
      Python

      Aggregation Queries

      def get_team_statistics():
    with Session(engine) as session:
        statement = select(
            Team.name,
            func.count(Hero.id).label("hero_count"),
            func.avg(Hero.age).label("average_age"),
            func.max(Hero.age).label("oldest_hero_age")
        ).select_from(
            Team
        ).join(
            Hero, Team.id == Hero.team_id, isouter=True
        ).group_by(
            Team.id, Team.name
        )

        results = session.exec(statement).all()
        return [
            {
                "team_name": result.name,
                "hero_count": result.hero_count,
                "average_age": result.average_age,
                "oldest_hero_age": result.oldest_hero_age
            }
            for result in results
        ]
      Python

      Raw SQL Queries

      def execute_raw_query():
    with Session(engine) as session:
        result = session.exec(
            text("SELECT name, age FROM hero WHERE age > :min_age"),
            {"min_age": 18}
        )
        return result.all()
      Python

      8. FastAPI Integration

      API Architecture

      graph TD
    A[FastAPI Router] --> B[Dependency Injection]
    B --> C[Database Session]
    C --> D[SQLModel Operations]
    D --> E[Response Models]

    F[Request] --> G[Validation]
    G --> H[Pydantic Model]
    H --> I[SQLModel]

      Setting up FastAPI with SQLModel

      # main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session
from typing import List
from models import Hero, HeroCreate, HeroRead, HeroUpdate
from database import get_session, create_db_and_tables

app = FastAPI(title="Hero API", version="1.0.0")

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

# CRUD Endpoints
@app.post("/heroes/", response_model=HeroRead)
def create_hero(
    hero: HeroCreate, 
    session: Session = Depends(get_session)
):
    db_hero = Hero.from_orm(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 = 100,
    session: Session = Depends(get_session)
):
    statement = select(Hero).offset(skip).limit(limit)
    heroes = session.exec(statement).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")

    hero_data = hero_update.dict(exclude_unset=True)
    for field, value in hero_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 {"message": "Hero deleted successfully"}
      Python

      Request/Response Models

      # models.py - API Models
class HeroBase(SQLModel):
    name: str = Field(min_length=1, max_length=50)
    age: Optional[int] = Field(default=None, ge=0, le=200)
    secret_name: str

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

class HeroCreate(HeroBase):
    pass

class HeroRead(HeroBase):
    id: int

class HeroUpdate(SQLModel):
    name: Optional[str] = Field(default=None, min_length=1, max_length=50)
    age: Optional[int] = Field(default=None, ge=0, le=200)
    secret_name: Optional[str] = None
      Python

      9. Testing

      Testing Strategy

      graph TD
    A[Unit Tests] --> B[Model Validation]
    A --> C[Database Operations]

    D[Integration Tests] --> E[API Endpoints]
    D --> F[Database Transactions]

    G[Test Database] --> H[In-Memory SQLite]
    G --> I[Test Fixtures]

      Setting up Test Environment

      # conftest.py
import pytest
from sqlmodel import SQLModel, create_engine, Session
from sqlmodel.pool import StaticPool
from fastapi.testclient import TestClient
from main import app, get_session

@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()
      Python

      Model Tests

      # test_models.py
def test_hero_creation():
    hero = Hero(
        name="Test Hero",
        secret_name="Test Secret",
        age=25
    )
    assert hero.name == "Test Hero"
    assert hero.age == 25

def test_hero_validation():
    with pytest.raises(ValidationError):
        Hero(
            name="",  # Invalid: empty name
            secret_name="Test",
            age=-1  # Invalid: negative age
        )
      Python

      Database Tests

      # test_database.py
def test_create_hero(session: Session):
    hero = Hero(name="Test Hero", secret_name="Secret")
    session.add(hero)
    session.commit()
    session.refresh(hero)

    assert hero.id is not None
    assert hero.name == "Test Hero"

def test_read_hero(session: Session):
    hero = Hero(name="Test Hero", secret_name="Secret")
    session.add(hero)
    session.commit()

    retrieved_hero = session.get(Hero, hero.id)
    assert retrieved_hero is not None
    assert retrieved_hero.name == "Test Hero"
      Python

      API Tests

      # test_api.py
def test_create_hero_api(client: TestClient):
    hero_data = {
        "name": "Test Hero",
        "secret_name": "Test Secret",
        "age": 25
    }
    response = client.post("/heroes/", json=hero_data)
    assert response.status_code == 200

    data = response.json()
    assert data["name"] == hero_data["name"]
    assert "id" in data

def test_read_heroes_api(client: TestClient):
    response = client.get("/heroes/")
    assert response.status_code == 200
    assert isinstance(response.json(), list)
      Python

      10. Performance Optimization

      Query Optimization Strategies

      graph TD
    A[Query Optimization] --> B[Eager Loading]
    A --> C[Lazy Loading]
    A --> D[Select Loading]
    A --> E[Batch Loading]

    F[Database Optimization] --> G[Indexes]
    F --> H[Connection Pooling]
    F --> I[Query Caching]

      Eager Loading

      # Avoid N+1 queries
def get_teams_with_heroes_optimized():
    with Session(engine) as session:
        statement = select(Team).options(selectinload(Team.heroes))
        teams = session.exec(statement).all()

        # Now accessing heroes won't trigger additional queries
        for team in teams:
            print(f"Team: {team.name}")
            for hero in team.heroes:  # No additional query
                print(f"  Hero: {hero.name}")

        return teams
      Python

      Connection Pooling

      # database.py - Production setup
from sqlmodel import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20,
    pool_recycle=3600,
    pool_pre_ping=True
)
      Python

      Batch Operations

      def bulk_create_heroes(heroes_data: List[dict]):
    with Session(engine) as session:
        heroes = [Hero(**data) for data in heroes_data]
        session.add_all(heroes)
        session.commit()
        return heroes

def bulk_update_heroes(updates: List[dict]):
    with Session(engine) as session:
        session.bulk_update_mappings(Hero, updates)
        session.commit()
      Python

      Caching Strategies

      from functools import lru_cache
import redis

# Simple memory cache
@lru_cache(maxsize=100)
def get_hero_cached(hero_id: int):
    with Session(engine) as session:
        return session.get(Hero, hero_id)

# Redis cache
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_hero_with_redis_cache(hero_id: int):
    cache_key = f"hero:{hero_id}"
    cached_hero = redis_client.get(cache_key)

    if cached_hero:
        return Hero.parse_raw(cached_hero)

    with Session(engine) as session:
        hero = session.get(Hero, hero_id)
        if hero:
            redis_client.setex(
                cache_key, 
                3600,  # 1 hour
                hero.json()
            )
        return hero
      Python

      11. Production Deployment

      Deployment Architecture

      graph TB
    A[Load Balancer] --> B[FastAPI App 1]
    A --> C[FastAPI App 2]
    A --> D[FastAPI App N]

    B --> E[Database Pool]
    C --> E
    D --> E

    E --> F[PostgreSQL Master]
    E --> G[PostgreSQL Replica]

    H[Redis Cache] --> B
    H --> C
    H --> D

      Environment Configuration

      # config.py
from pydantic import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    database_url: str
    redis_url: Optional[str] = None
    secret_key: str
    debug: bool = False

    class Config:
        env_file = ".env"

settings = Settings()
      Python

      Database Migrations with Alembic

      # Install Alembic
pip install alembic

# Initialize
alembic init alembic

# Create migration
alembic revision --autogenerate -m "Initial migration"

# Apply migration
alembic upgrade head
      Bash

      Docker Deployment

      # Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
      Dockerfile
      # docker-compose.yml
version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/heroesdb
    depends_on:
      - db
      - redis

  db:
    image: postgres:15
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: heroesdb
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  postgres_data:
      YAML

      12. Best Practices

      Code Organization

      graph TD
    A[Project Structure] --> B[models/]
    A --> C[routers/]
    A --> D[services/]
    A --> E[dependencies/]

    B --> F[base.py]
    B --> G[hero.py]
    B --> H[team.py]

    C --> I[heroes.py]
    C --> J[teams.py]

    D --> K[hero_service.py]
    D --> L[team_service.py]

      Security Best Practices

      # security.py
from passlib.context import CryptContext
from sqlmodel import SQLModel, Field
from typing import Optional

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(unique=True, index=True)
    email: str = Field(unique=True, index=True)
    hashed_password: str
    is_active: bool = Field(default=True)

    def verify_password(self, password: str) -> bool:
        return pwd_context.verify(password, self.hashed_password)

    @classmethod
    def hash_password(cls, password: str) -> str:
        return pwd_context.hash(password)
      Python

      Error Handling

      # exceptions.py
class HeroNotFoundError(Exception):
    def __init__(self, hero_id: int):
        self.hero_id = hero_id
        super().__init__(f"Hero with id {hero_id} not found")

class DatabaseError(Exception):
    pass

# error_handlers.py
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse

@app.exception_handler(HeroNotFoundError)
async def hero_not_found_handler(request: Request, exc: HeroNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Hero with id {exc.hero_id} not found"}
    )
      Python

      Logging

      # logging_config.py
import logging
from sqlmodel import Session

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

logger = logging.getLogger(__name__)

def log_database_operation(operation: str, model: str, record_id: int = None):
    logger.info(f"Database {operation} on {model}" + 
                (f" with id {record_id}" if record_id else ""))
      Python

      Data Validation

      from pydantic import validator, root_validator

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(min_length=1, max_length=100)
    age: Optional[int] = Field(default=None, ge=0, le=300)
    power_level: int = Field(ge=1, le=10)

    @validator('name')
    def validate_name(cls, v):
        if not v.strip():
            raise ValueError('Name cannot be empty or whitespace')
        return v.title()

    @root_validator
    def validate_hero_constraints(cls, values):
        age = values.get('age')
        power_level = values.get('power_level')

        if age and age < 18 and power_level > 5:
            raise ValueError('Minors cannot have power level > 5')

        return values
      Python

      Conclusion

      SQLModel provides a powerful, type-safe way to work with databases in Python applications. By combining the best of SQLAlchemy and Pydantic, it offers:

      • Developer Experience: Type hints and IDE support
      • Performance: Efficient database operations
      • Maintainability: Single source of truth for models
      • Integration: Seamless FastAPI development

      Key Takeaways

      1. Always use type hints for better code quality
      2. Leverage relationships for complex data models
      3. Implement proper error handling and logging
      4. Use dependency injection for testable code
      5. Optimize queries for production performance
      6. Follow security best practices
      7. Test thoroughly with proper fixtures

      Next Steps

      • Explore async SQLModel with databases
      • Learn about advanced SQLAlchemy features
      • Implement microservices architecture
      • Study database design patterns
      • Practice with real-world projects

      Remember: Start simple, iterate, and always prioritize code clarity and maintainability.

      Discover more from Altgr Blog

      Subscribe to get the latest posts sent to your email.

      Related Posts

      Leave a Reply

      Your email address will not be published. Required fields are marked *