SQLMODEL - ORM based on Pydan

Introduction

SQLAlchemy and Pydantic are two powerful tools for Python developers: one for working with databases, the other for data validation. However, a gap has always existed between them. The solution to this problem is SQLModel — an ORM built on top of Pydantic that combines the best of both worlds.

SQLModel allows you to use typed Pydantic models as SQLAlchemy ORM tables. This simplifies development, speeds up API creation, and eliminates code duplication between database models and serializers.

What Is SQLModel and Why You Need It

SQLModel is a modern ORM library for Python created by Sebastián Ramírez (the creator of FastAPI) in 2021. It solves one of the biggest challenges in web development — code duplication between database models and data‑validation schemas.

Main Benefits of SQLModel

SQLModel provides developers with a single interface for database interaction and data validation. Key advantages include:

• Unified data model: one model serves both the database and the API
• Automatic validation: built‑in Pydantic validation
• Typing: full support for Python type hints and IDE autocompletion
• Compatibility: works with existing SQLAlchemy projects
• Asynchrony: out‑of‑the‑box async/await support

Installation and Setup

Basic Installation

pip install sqlmodel

Installation with Extra Dependencies

For PostgreSQL support:

pip install sqlmodel[postgresql]

For asynchronous operation:

pip install sqlmodel[async]

Full installation with all optional dependencies:

pip install "sqlmodel[postgresql,async]"

Connecting to a Database

from sqlmodel import SQLModel, create_engine

# SQLite (development)
engine = create_engine("sqlite:///database.db")

# PostgreSQL
engine = create_engine("postgresql://user:password@localhost/dbname")

# MySQL
engine = create_engine("mysql+pymysql://user:password@localhost/dbname")

Fundamentals of Working with SQLModel

Creating a Basic Model

from sqlmodel import SQLModel, Field
from typing import Optional

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    email: str = Field(unique=True)
    age: int = Field(ge=0, le=150)
    is_active: bool = Field(default=True)

Field Parameters

SQLModel uses an extended Pydantic Field with additional capabilities:

from datetime import datetime

class Product(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(max_length=100, index=True)
    price: float = Field(gt=0, description="Product price")
    category_id: int = Field(foreign_key="category.id")
    created_at: datetime = Field(default_factory=datetime.now)

Creating Tables in the Database

from sqlmodel import SQLModel, create_engine

engine = create_engine("sqlite:///database.db")

# Create all tables
SQLModel.metadata.create_all(engine)

# Create a specific table
User.metadata.create_all(engine)

CRUD Operations with SQLModel

Creating Records

from sqlmodel import Session

# Create a single record
user = User(name="Ivan", email="ivan@example.com", age=25)

with Session(engine) as session:
    session.add(user)
    session.commit()
    session.refresh(user)  # Retrieve the generated ID
    print(f"Created user with ID: {user.id}")

# Create multiple records
users = [
    User(name="Alexey", email="alex@example.com", age=30),
    User(name="Maria", email="maria@example.com", age=28)
]

with Session(engine) as session:
    session.add_all(users)
    session.commit()

Reading Data

from sqlmodel import select

with Session(engine) as session:
    # Retrieve all users
    statement = select(User)
    users = session.exec(statement).all()
    
    # Retrieve a user by ID
    user = session.get(User, 1)
    
    # Retrieve the first user older than 25
    statement = select(User).where(User.age > 25)
    user = session.exec(statement).first()
    
    # Filter by name containing "Ivan"
    statement = select(User).where(User.name.contains("Ivan"))
    users = session.exec(statement).all()

Updating Data

with Session(engine) as session:
    # Fetch and update a single record
    user = session.get(User, 1)
    if user:
        user.age = 26
        user.email = "new_email@example.com"
        session.add(user)
        session.commit()
    
    # Bulk update
    statement = select(User).where(User.age < 18)
    users = session.exec(statement).all()
    for user in users:
        user.is_active = False
        session.add(user)
    session.commit()

Deleting Data

with Session(engine) as session:
    # Delete a specific record
    user = session.get(User, 1)
    if user:
        session.delete(user)
        session.commit()
    
    # Bulk delete
    statement = select(User).where(User.is_active == False)
    users = session.exec(statement).all()
    for user in users:
        session.delete(user)
    session.commit()

Advanced Queries and Filtering

Filter Operators

from sqlmodel import select, and_, or_

with Session(engine) as session:
    # Exact match
    statement = select(User).where(User.name == "Ivan")
    
    # Comparison
    statement = select(User).where(User.age >= 18)
    
    # Substring search
    statement = select(User).where(User.email.contains("@gmail.com"))
    
    # IN list
    statement = select(User).where(User.name.in_(["Ivan", "Maria"]))
    
    # Combined conditions
    statement = select(User).where(
        and_(User.age >= 18, User.is_active == True)
    )
    
    # OR condition
    statement = select(User).where(
        or_(User.age < 18, User.age > 65)
    )

Sorting and Limits

# Sorting
statement = select(User).order_by(User.name)
statement = select(User).order_by(User.age.desc())

# Limit and offset
statement = select(User).limit(10).offset(20)

# Combined example
statement = select(User).where(User.is_active == True).order_by(User.name).limit(5)

Data Aggregation

from sqlmodel import func

with Session(engine) as session:
    # Count rows
    statement = select(func.count(User.id))
    count = session.exec(statement).one()
    
    # Average age
    statement = select(func.avg(User.age))
    avg_age = session.exec(statement).one()
    
    # Group by age
    statement = select(User.age, func.count(User.id)).group_by(User.age)
    results = session.exec(statement).all()

Working with Table Relationships

Defining Relationships

from sqlmodel import Relationship
from typing import Optional, List

class Category(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    
    # Reverse relationship
    products: List["Product"] = Relationship(back_populates="category")

class Product(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    price: float
    category_id: int = Field(foreign_key="category.id")
    
    # Forward relationship
    category: Optional[Category] = Relationship(back_populates="products")

Working with Related Data

with Session(engine) as session:
    # Create related records
    category = Category(name="Electronics")
    session.add(category)
    session.commit()
    session.refresh(category)
    
    product = Product(
        name="Smartphone",
        price=25000,
        category_id=category.id
    )
    session.add(product)
    session.commit()
    
    # Retrieve related data
    statement = select(Product).where(Product.id == 1)
    product = session.exec(statement).first()
    print(f"Product: {product.name}, Category: {product.category.name}")

Many‑to‑Many Relationships

class UserProductLink(SQLModel, table=True):
    user_id: int = Field(foreign_key="user.id", primary_key=True)
    product_id: int = Field(foreign_key="product.id", primary_key=True)

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    
    products: List[Product] = Relationship(
        back_populates="users",
        link_model=UserProductLink
    )

class Product(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    
    users: List[User] = Relationship(
        back_populates="products",
        link_model=UserProductLink
    )

Asynchronous Work with SQLModel

Setting Up an Async Engine

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlmodel import select

# Async SQLite engine
engine = create_async_engine("sqlite+aiosqlite:///database.db")

# Async PostgreSQL engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/dbname")

Async Operations

async def create_user(name: str, email: str):
    async with AsyncSession(engine) as session:
        user = User(name=name, email=email)
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

async def get_users():
    async with AsyncSession(engine) as session:
        statement = select(User)
        result = await session.exec(statement)
        return result.all()

async def update_user(user_id: int, new_name: str):
    async with AsyncSession(engine) as session:
        user = await session.get(User, user_id)
        if user:
            user.name = new_name
            session.add(user)
            await session.commit()
            await session.refresh(user)
        return user

Integration with FastAPI

Basic Integration

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
from typing import List

app = FastAPI()

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

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

@app.get("/users/", response_model=List[User])
def read_users(session: Session = Depends(get_session)):
    users = session.exec(select(User)).all()
    return users

@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, session: Session = Depends(get_session)):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Separating API Schemas

class UserBase(SQLModel):
    name: str
    email: str
    age: int

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    is_active: bool = Field(default=True)

class UserCreate(UserBase):
    pass

class UserRead(UserBase):
    id: int
    is_active: bool

class UserUpdate(SQLModel):
    name: Optional[str] = None
    email: Optional[str] = None
    age: Optional[int] = None

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

Migrations with Alembic

Setting Up Alembic

pip install alembic
alembic init alembic

Configuring env.py

# alembic/env.py
from alembic import context
from sqlmodel import SQLModel
from myapp.models import User, Product, Category  # Import all models

target_metadata = SQLModel.metadata

def run_migrations_online():
    connectable = engine
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )
        with context.begin_transaction():
            context.run_migrations()

Creating and Applying Migrations

# Generate a migration
alembic revision --autogenerate -m "Add user table"

# Apply the migration
alembic upgrade head

# Roll back a migration
alembic downgrade -1

Data Validation

Built‑in Pydantic Validation

from pydantic import validator, root_validator
from typing import Optional

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(min_length=2, max_length=50)
    email: str = Field(regex=r'^[^@]+@[^@]+\.[^@]+$')
    age: int = Field(ge=0, le=150)
    password: str = Field(min_length=8)
    
    @validator('email')
    def validate_email(cls, v):
        if not v.endswith('@example.com'):
            raise ValueError('Only emails with the @example.com domain are allowed')
        return v
    
    @validator('name')
    def validate_name(cls, v):
        if any(char.isdigit() for char in v):
            raise ValueError('Name must not contain digits')
        return v.title()
    
    @root_validator
    def validate_password_strength(cls, values):
        password = values.get('password')
        if password and len(password) < 8:
            raise ValueError('Password must be at least 8 characters long')
        return values

Custom Validators

from pydantic import validator
import re

class Product(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    price: float
    sku: str = Field(unique=True)
    
    @validator('price')
    def validate_price(cls, v):
        if v <= 0:
            raise ValueError('Price must be positive')
        return round(v, 2)
    
    @validator('sku')
    def validate_sku(cls, v):
        if not re.match(r'^[A-Z]{3}-\\d{3}$', v):
            raise ValueError('SKU must follow the format ABC-123')
        return v

Testing with SQLModel

Setting Up a Test Database

import pytest
from sqlmodel import SQLModel, create_engine, Session
from sqlalchemy.pool import StaticPool

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

Testing CRUD Operations

def test_create_user(session: Session):
    user = User(name="Test User", email="test@example.com", age=25)
    session.add(user)
    session.commit()
    session.refresh(user)
    
    assert user.id is not None
    assert user.name == "Test User"
    assert user.email == "test@example.com"

def test_read_user(session: Session):
    user = User(name="Test User", email="test@example.com", age=25)
    session.add(user)
    session.commit()
    session.refresh(user)
    
    retrieved_user = session.get(User, user.id)
    assert retrieved_user is not None
    assert retrieved_user.name == "Test User"

def test_update_user(session: Session):
    user = User(name="Test User", email="test@example.com", age=25)
    session.add(user)
    session.commit()
    session.refresh(user)
    
    user.name = "Updated Name"
    session.add(user)
    session.commit()
    
    updated_user = session.get(User, user.id)
    assert updated_user.name == "Updated Name"

Performance and Optimization

Eager Loading

from sqlmodel import select
from sqlalchemy.orm import selectinload

# Load related data in a single query
statement = select(User).options(selectinload(User.products))
users = session.exec(statement).all()

# Each user now has products pre‑loaded
for user in users:
    print(f"User {user.name} has {len(user.products)} products")

Bulk Operations

def bulk_create_users(users_data: List[dict]):
    with Session(engine) as session:
        users = [User(**data) for data in users_data]
        session.add_all(users)
        session.commit()
        return users

def bulk_update_users(updates: List[dict]):
    with Session(engine) as session:
        session.bulk_update_mappings(User, updates)
        session.commit()

Indexes and Optimization

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)          # Simple index
    email: str = Field(unique=True)        # Unique index
    age: int
    city: str = Field(index=True)
    
    # Composite index via __table_args__
    __table_args__ = (
        Index('idx_age_city', 'age', 'city'),
    )

Key SQLModel Methods and Functions

Integration with Analytics Tools

Working with Pandas

import pandas as pd
from sqlmodel import select

def users_to_dataframe(session: Session) -> pd.DataFrame:
    statement = select(User)
    users = session.exec(statement).all()
    
    # Convert to list of dictionaries
    users_data = [user.dict() for user in users]
    
    # Create DataFrame
    df = pd.DataFrame(users_data)
    return df

# Usage example
with Session(engine) as session:
    df = users_to_dataframe(session)
    print(df.describe())
    
    # Simple analysis
    age_stats = df.groupby('age').size()
    print(age_stats)

Exporting Data

def export_users_to_csv(session: Session, filename: str):
    df = users_to_dataframe(session)
    df.to_csv(filename, index=False)

def export_users_to_json(session: Session, filename: str):
    statement = select(User)
    users = session.exec(statement).all()
    
    users_data = [user.dict() for user in users]
    
    import json
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(users_data, f, ensure_ascii=False, indent=2)

Comparison with Other ORMs

Frequently Asked Questions

What is SQLModel and how does it differ from SQLAlchemy?

SQLModel is an ORM library that merges SQLAlchemy and Pydantic into a single syntax. Key differences: built‑in data validation, automatic API schema generation, full typing support, and tighter FastAPI integration.

Can I use SQLModel without FastAPI?

Yes, SQLModel is completely independent and can be used in any Python project. It does not depend on FastAPI, although it was created by the same author for perfect compatibility.

Does SQLModel support asynchronous operations?

Yes, SQLModel supports asynchrony via SQLAlchemy 2.0 and an async engine. You can use AsyncSession for async database interactions.

How do I create a table from a model?

Use SQLModel.metadata.create_all(engine) to create all tables, or Model.metadata.create_all(engine) to create a specific table.

How do I implement a foreign key in SQLModel?

Use Field(foreign_key="tablename.columnname"). Example: user_id: int = Field(foreign_key="user.id").

Does SQLModel work with PostgreSQL and other databases?

Yes, SQLModel supports every DBMS that SQLAlchemy supports: PostgreSQL, MySQL, SQLite, Oracle, Microsoft SQL Server, and more.

How are validation errors handled?

SQLModel validates data automatically when creating objects. Validation errors raise Pydantic ValidationError exceptions, which you can catch and handle in your code.

Can I use SQLModel with an existing database?

Absolutely. SQLModel is compatible with existing databases. You can define models that map to existing tables and gradually migrate your project.

How can I optimize query performance?

Use indexes via Field(index=True), eager loading for related data, bulk operations for large datasets, and well‑structured filtered queries.

Does SQLModel support migrations?

SQLModel uses Alembic for migrations, just like SQLAlchemy. You can generate and apply migration scripts to evolve your database schema.