SQLModel: ORM Modern Python untuk Aplikasi AI yang Type-Safe

Dalam pengembangan aplikasi AI/ML, pengelolaan data di database adalah komponen krusial. Dari menyimpan hasil eksperimen, mengelola model registry, hingga tracking dataset versioning, semuanya membutuhkan interaksi database yang efisien dan type-safe. SQLModel, yang dibuat oleh Sebastián Ramírez (kreator FastAPI), menggabungkan kekuatan SQLAlchemy dan Pydantic untuk memberikan pengalaman ORM terbaik di Python.

Dalam tutorial ini, kita akan mempelajari SQLModel secara mendalam dan membangun sistem ML experiment tracking dan model registry yang siap produksi.

Apa Itu SQLModel?

SQLModel adalah library Python yang menggabungkan SQLAlchemy (ORM paling populer di Python) dengan Pydantic (library validasi data). Hasilnya adalah ORM yang:

• Type-safe: Full type hints dan autocompletion di IDE
• Validasi otomatis: Data divalidasi sebelum masuk database
• Kompatibel dengan FastAPI: Integrasi seamless karena kreator yang sama
• Sederhana: Satu class untuk model database DAN schema API
• Powerful: Akses penuh ke fitur SQLAlchemy jika dibutuhkan

Instalasi dan Setup

Instalasi SQLModel

pip install sqlmodel

Untuk database spesifik:

# PostgreSQL
pip install sqlmodel psycopg2-binary

MySQL
pip install sqlmodel pymysql

Async support
pip install sqlmodel aiosqlite asyncpg

Verifikasi
python -c "import sqlmodel; print(sqlmodel.version)"

Setup Database Engine

from sqlmodel import createengine, SQLModel

SQLite (development)
sqliteurl = "sqlite:///./mltracking.db"

engine = createengine(sqliteurl, echo=True)


PostgreSQL (production)
postgresurl = "postgresql://user:password@localhost:5432/mltracking"

engine = createengine(postgresurl, echo=False, poolsize=20, maxoverflow=10)


Buat semua tabel
SQLModel.metadata.createall(engine)

Definisi Model

Model Dasar

from sqlmodel import Field, SQLModel, Session, select
from typing import Optional
from datetime import datetime
import uuid

class Experiment(SQLModel, table=True):
    """Model untuk ML Experiment"""
    id: Optional[int] = Field(default=None, primarykey=True)

    name: str = Field(index=True, minlength=1, maxlength=255)

    description: Optional[str] = Field(default=None, maxlength=1000)
    datasetname: str = Field(index=True)

    algorithm: str
    hyperparameters: Optional[str] = Field(default=None)  # JSON string
    accuracy: Optional[float] = Field(default=None, ge=0.0, le=1.0)
    loss: Optional[float] = Field(default=None, ge=0.0)
    status: str = Field(default="created", index=True)
    createdat: datetime = Field(defaultfactory=datetime.utcnow)

    updatedat: Optional[datetime] = Field(default=None)
    createdby: str = Field(default="system")


class MLModel(SQLModel, table=True):
    """Model untuk Model Registry"""
    tablename = "mlmodels"

    id: Optional[int] = Field(default=None, primarykey=True)

    name: str = Field(index=True, unique=True)
    version: str = Field(default="1.0.0")
    framework: str  # pytorch, tensorflow, sklearn, etc.
    modelpath: str
    filesizemb: Optional[float] = Field(default=None)
    inputschema: Optional[str] = Field(default=None)

    outputschema: Optional[str] = Field(default=None)
    isactive: bool = Field(default=True, index=True)

    createdat: datetime = Field(defaultfactory=datetime.utcnow)


class Dataset(SQLModel, table=True):
    """Model untuk Dataset Registry"""
    id: Optional[int] = Field(default=None, primarykey=True)
    name: str = Field(index=True)
    version: str = Field(default="1.0")
    filepath: str

    rowcount: Optional[int] = Field(default=None)
    columncount: Optional[int] = Field(default=None)

    fileformat: str = Field(default="csv")
    description: Optional[str] = Field(default=None)
    createdat: datetime = Field(defaultfactory=datetime.utcnow)

Model dengan Read dan Create Schemas

SQLModel memungkinkan pembuatan schema terpisah untuk read dan create:

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

Base model (shared fields)
class ExperimentBase(SQLModel):
    name: str = Field(minlength=1, maxlength=255)
    description: Optional[str] = None
    datasetname: str

    algorithm: str
    hyperparameters: Optional[str] = None

Create schema (untuk input API)
class ExperimentCreate(ExperimentBase):
    pass

Update schema
class ExperimentUpdate(SQLModel):
    name: Optional[str] = None
    description: Optional[str] = None
    accuracy: Optional[float] = None
    loss: Optional[float] = None
    status: Optional[str] = None

Database model (tabel)
class Experiment(ExperimentBase, table=True):
    id: Optional[int] = Field(default=None, primarykey=True)
    accuracy: Optional[float] = Field(default=None, ge=0.0, le=1.0)
    loss: Optional[float] = Field(default=None, ge=0.0)
    status: str = Field(default="created", index=True)
    createdat: datetime = Field(defaultfactory=datetime.utcnow)
    updatedat: Optional[datetime] = None


Read schema (untuk response API)
class ExperimentRead(ExperimentBase):
    id: int
    accuracy: Optional[float]
    loss: Optional[float]
    status: str
    createdat: datetime

Operasi CRUD

Create (Insert Data)

from sqlmodel import Session

def createexperiment(engine, experimentdata: ExperimentCreate) -> Experiment:
    """Membuat experiment baru"""
    experiment = Experiment.modelvalidate(experimentdata)
    with Session(engine) as session:
        session.add(experiment)
        session.commit()
        session.refresh(experiment)
        return experiment

Contoh penggunaan
newexp = ExperimentCreate(

    name="ResNet50 Transfer Learning",
    description="Fine-tuning ResNet50 pada dataset custom",
    datasetname="chest-xray-v2",
    algorithm="ResNet50",
    hyperparameters='{"lr": 0.001, "epochs": 50, "batchsize": 32}',

)
experiment = createexperiment(engine, newexp)

print(f"Experiment created: ID={experiment.id}")

Batch Insert

def createexperimentsbatch(engine, experiments: list[ExperimentCreate]) -> list[Experiment]:
    """Insert banyak experiments sekaligus"""
    dbexperiments = [Experiment.modelvalidate(exp) for exp in experiments]

    with Session(engine) as session:
        session.addall(dbexperiments)

        session.commit()
        for exp in dbexperiments:
            session.refresh(exp)
        return dbexperiments

Read (Query Data)

from sqlmodel import Session, select

def getexperimentbyid(engine, experimentid: int) -> Optional[Experiment]:

    """Ambil experiment berdasarkan ID"""
    with Session(engine) as session:
        return session.get(Experiment, experimentid)

def getallexperiments(engine, skip: int = 0, limit: int = 100) -> list[Experiment]:
    """Ambil semua experiments dengan pagination"""
    with Session(engine) as session:
        statement = select(Experiment).offset(skip).limit(limit)
        return session.exec(statement).all()

def getexperimentsbystatus(engine, status: str) -> list[Experiment]:

    """Filter experiments berdasarkan status"""
    with Session(engine) as session:
        statement = select(Experiment).where(Experiment.status == status)
        return session.exec(statement).all()

def getbestexperiments(engine, topn: int = 10) -> list[Experiment]:
    """Ambil experiments dengan akurasi tertinggi"""
    with Session(engine) as session:
        statement = (
            select(Experiment)
            .where(Experiment.accuracy.isnot(None))

            .orderby(Experiment.accuracy.desc())
            .limit(topn)

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

def searchexperiments(engine, query: str) -> list[Experiment]:
    """Cari experiments berdasarkan nama atau algoritma"""
    with Session(engine) as session:
        statement = select(Experiment).where(
            (Experiment.name.contains(query)) |
            (Experiment.algorithm.contains(query))
        )
        return session.exec(statement).all()

Update

def updateexperiment(
    engine, experimentid: int, updatedata: ExperimentUpdate

) -> Optional[Experiment]:
    """Update experiment"""
    with Session(engine) as session:
        experiment = session.get(Experiment, experimentid)
        if not experiment:
            return None

        updatedict = updatedata.modeldump(excludeunset=True)
        for key, value in updatedict.items():

            setattr(experiment, key, value)

        experiment.updatedat = datetime.utcnow()
        session.add(experiment)
        session.commit()
        session.refresh(experiment)
        return experiment

Contoh: update hasil training
update = ExperimentUpdate(accuracy=0.95, loss=0.12, status="completed")
updated = updateexperiment(engine, experimentid=1, updatedata=update)

Delete

def deleteexperiment(engine, experimentid: int) -> bool:
    """Hapus experiment"""
    with Session(engine) as session:
        experiment = session.get(Experiment, experimentid)
        if not experiment:
            return False
        session.delete(experiment)
        session.commit()
        return True

def deleteexperimentsbystatus(engine, status: str) -> int:

    """Hapus semua experiments dengan status tertentu"""
    with Session(engine) as session:
        statement = select(Experiment).where(Experiment.status == status)
        experiments = session.exec(statement).all()
        count = len(experiments)
        for exp in experiments:
            session.delete(exp)
        session.commit()
        return count

Relationships

SQLModel mendukung relasi antar tabel sama seperti SQLAlchemy:

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

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primarykey=True)
    name: str = Field(index=True, unique=True)
    description: Optional[str] = None

    # Relationship: satu team punya banyak experiments
    experiments: list["Experiment"] = Relationship(backpopulates="team")

    members: list["TeamMember"] = Relationship(backpopulates="team")

class TeamMember(SQLModel, table=True):
    tablename = "teammembers"


    id: Optional[int] = Field(default=None, primarykey=True)
    name: str
    email: str = Field(unique=True)
    role: str = Field(default="member")

    teamid: Optional[int] = Field(default=None, foreignkey="team.id")
    team: Optional[Team] = Relationship(backpopulates="members")


class Experiment(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primarykey=True)
    name: str = Field(index=True)
    algorithm: str
    accuracy: Optional[float] = None
    status: str = Field(default="created")
    createdat: datetime = Field(defaultfactory=datetime.utcnow)

    # Foreign key ke team
    teamid: Optional[int] = Field(default=None, foreignkey="team.id")
    team: Optional[Team] = Relationship(backpopulates="experiments")


    # Relationship ke metrics
    metrics: list["ExperimentMetric"] = Relationship(backpopulates="experiment")

class ExperimentMetric(SQLModel, table=True):
    tablename = "experimentmetrics"


    id: Optional[int] = Field(default=None, primarykey=True)
    epoch: int
    trainloss: float

    valloss: float
    trainaccuracy: float

    valaccuracy: float

    experimentid: int = Field(foreignkey="experiment.id")
    experiment: Optional[Experiment] = Relationship(backpopulates="metrics")

Query dengan Relationships

def getteamwithexperiments(engine, teamid: int) -> Optional[Team]:
    """Ambil team beserta semua experiments"""
    with Session(engine) as session:
        statement = select(Team).where(Team.id == teamid)
        team = session.exec(statement).first()
        if team:
            # Akses experiments (lazy loaded)
             = team.experiments

        return team

def getexperimentwithmetrics(engine, experimentid: int):

    """Ambil experiment beserta semua metrics per epoch"""
    with Session(engine) as session:
        statement = select(Experiment).where(Experiment.id == experimentid)
        experiment = session.exec(statement).first()
        if experiment:
             = experiment.metrics

        return experiment

Integrasi dengan FastAPI

SQLModel dirancang untuk bekerja sempurna dengan FastAPI:

from fastapi import FastAPI, HTTPException, Depends, Query
from sqlmodel import Session, select, createengine, SQLModel
from typing import Optional

DATABASEURL = "postgresql://user:password@localhost:5432/mltracking"
engine = createengine(DATABASEURL, poolsize=20)


app = FastAPI(title="ML Experiment Tracker API")

def getsession():
    """Dependency untuk database session"""
    with Session(engine) as session:
        yield session

@app.onevent("startup")

def onstartup():
    SQLModel.metadata.createall(engine)


@app.post("/experiments/", responsemodel=ExperimentRead)
def createexperiment(

    experiment: ExperimentCreate,
    session: Session = Depends(getsession),
):
    dbexperiment = Experiment.modelvalidate(experiment)
    session.add(dbexperiment)

    session.commit()
    session.refresh(dbexperiment)
    return dbexperiment


@app.get("/experiments/", responsemodel=list[ExperimentRead])
def listexperiments(

    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=20, le=100),
    status: Optional[str] = None,
    algorithm: Optional[str] = None,
    session: Session = Depends(getsession),
):
    statement = select(Experiment)
    if status:
        statement = statement.where(Experiment.status == status)
    if algorithm:
        statement = statement.where(Experiment.algorithm == algorithm)
    statement = statement.offset(skip).limit(limit)
    return session.exec(statement).all()

@app.get("/experiments/{experimentid}", responsemodel=ExperimentRead)
def getexperiment(experimentid: int, session: Session = Depends(getsession)):

    experiment = session.get(Experiment, experimentid)
    if not experiment:
        raise HTTPException(statuscode=404, detail="Experiment not found")

    return experiment

@app.patch("/experiments/{experimentid}", responsemodel=ExperimentRead)

def updateexperiment(
    experimentid: int,

    updatedata: ExperimentUpdate,
    session: Session = Depends(getsession),

):
    experiment = session.get(Experiment, experimentid)
    if not experiment:
        raise HTTPException(statuscode=404, detail="Experiment not found")


    for key, value in updatedata.modeldump(excludeunset=True).items():
        setattr(experiment, key, value)

    experiment.updatedat = datetime.utcnow()

    session.add(experiment)
    session.commit()
    session.refresh(experiment)
    return experiment

@app.delete("/experiments/{experimentid}")
def deleteexperiment(experimentid: int, session: Session = Depends(getsession)):

    experiment = session.get(Experiment, experimentid)
    if not experiment:
        raise HTTPException(statuscode=404, detail="Experiment not found")

    session.delete(experiment)
    session.commit()
    return {"message": "Experiment deleted", "id": experimentid}

@app.get("/experiments/best/", responsemodel=list[ExperimentRead])

def getbestexperiments(

    topn: int = Query(default=10, le=50),
    session: Session = Depends(getsession),

):
    statement = (
        select(Experiment)
        .where(Experiment.accuracy.isnot(None))
        .orderby(Experiment.accuracy.desc())

        .limit(topn)
    )
    return session.exec(statement).all()

Async Support

SQLModel mendukung operasi asynchronous untuk performa tinggi:

from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import createasyncengine, AsyncEngine
from sqlalchemy.orm import sessionmaker

Async engine
asyncengine = createasyncengine(

    "postgresql+asyncpg://user:password@localhost:5432/mltracking",
    echo=False,
    poolsize=20,

)

asyncsession = sessionmaker(asyncengine, class=AsyncSession, expireoncommit=False)

async def getasyncsession():
    async with asyncsession() as session:

        yield session

Async CRUD operations
from fastapi import FastAPI, Depends
from sqlmodel import select

app = FastAPI()

@app.post("/experiments/", responsemodel=ExperimentRead)
async def createexperimentasync(
    experiment: ExperimentCreate,
    session: AsyncSession = Depends(getasyncsession),
):
    dbexperiment = Experiment.modelvalidate(experiment)
    session.add(dbexperiment)

    await session.commit()
    await session.refresh(dbexperiment)
    return dbexperiment


@app.get("/experiments/", responsemodel=list[ExperimentRead])
async def listexperimentsasync(
    session: AsyncSession = Depends(getasyncsession),
):
    result = await session.exec(select(Experiment))
    return result.all()

Migration dengan Alembic

Alembic adalah tool standar untuk database migration di ekosistem SQLAlchemy/SQLModel:

# Install
pip install alembic

Inisialisasi
alembic init alembic

Konfigurasi alembic/env.py:

from sqlmodel import SQLModel
from app.models import Experiment, MLModel, Dataset, Team  # Import semua model

targetmetadata = SQLModel.metadata


def runmigrationsonline():

    connectable = enginefromconfig(

        config.getsection(config.configinisection),
        prefix="sqlalchemy.",
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            targetmetadata=targetmetadata,
        )
        with context.begintransaction():

            context.runmigrations()

Membuat dan menjalankan migration:

# Generate migration otomatis
alembic revision --autogenerate -m "add experiment metrics table"

Jalankan migration
alembic upgrade head

Rollback
alembic downgrade -1

Lihat history
alembic history

Query Building Lanjutan

from sqlmodel import Session, select, func, col, or, and

def advancedqueries(engine):

    with Session(engine) as session:
        # Aggregasi: rata-rata akurasi per algoritma
        statement = (
            select(
                Experiment.algorithm,
                func.count(Experiment.id).label("total"),
                func.avg(Experiment.accuracy).label("avgaccuracy"),
                func.max(Experiment.accuracy).label("bestaccuracy"),

            )
            .where(Experiment.status == "completed")
            .groupby(Experiment.algorithm)
            .orderby(func.avg(Experiment.accuracy).desc())

        )
        results = session.exec(statement).all()

        # Subquery: experiments yang di atas rata-rata
        avgsubquery = select(func.avg(Experiment.accuracy)).scalarsubquery()

        aboveavg = select(Experiment).where(
            Experiment.accuracy > avgsubquery

        )
        topexperiments = session.exec(aboveavg).all()


        # OR conditions
        statement = select(Experiment).where(
            or(
                Experiment.algorithm == "ResNet50",
                Experiment.algorithm == "EfficientNet",
            )
        )

        # LIKE search
        statement = select(Experiment).where(
            Experiment.name.like("%transfer%")
        )

        # IN clause
        algorithms = ["ResNet50", "VGG16", "EfficientNet"]
        statement = select(Experiment).where(
            Experiment.algorithm.in(algorithms)

        )

        # ORDER BY multiple columns
        statement = (
            select(Experiment)
            .orderby(Experiment.accuracy.desc(), Experiment.createdat.desc())

        )

        return results

Praktik: Sistem ML Experiment Tracking dan Model Registry

Berikut contoh lengkap sistem tracking yang siap produksi:

from fastapi import FastAPI, HTTPException, Depends, Query, UploadFile, File
from sqlmodel import Session, select, createengine, SQLModel, Field, Relationship
from typing import Optional
from datetime import datetime
import json
import os

Models
class MLModelBase(SQLModel):
    name: str = Field(index=True)
    version: str = Field(default="1.0.0")
    framework: str
    description: Optional[str] = None

class MLModelCreate(MLModelBase):
    pass

class MLModelDB(MLModelBase, table=True):
    tablename = "modelregistry"


    id: Optional[int] = Field(default=None, primarykey=True)
    modelpath: Optional[str] = None

    filesizemb: Optional[float] = None

    accuracy: Optional[float] = None
    isactive: bool = Field(default=True, index=True)
    isproduction: bool = Field(default=False)

    createdat: datetime = Field(defaultfactory=datetime.utcnow)

    metadatajson: Optional[str] = None

class MLModelRead(MLModelBase):
    id: int
    modelpath: Optional[str]

    accuracy: Optional[float]
    isactive: bool
    isproduction: bool

    createdat: datetime

Application
DATABASEURL = os.getenv("DATABASEURL", "sqlite:///./mlregistry.db")

engine = createengine(DATABASEURL)


app = FastAPI(title="ML Model Registry & Experiment Tracker")

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

@app.onevent("startup")

def startup():
    SQLModel.metadata.createall(engine)

@app.post("/models/register", responsemodel=MLModelRead)

def registermodel(
    modeldata: MLModelCreate,

    session: Session = Depends(getsession),
):
    """Registrasi model baru ke registry"""
    dbmodel = MLModelDB.modelvalidate(modeldata)

    session.add(dbmodel)
    session.commit()
    session.refresh(dbmodel)

    return dbmodel

@app.post("/models/{modelid}/promote")

def promotetoproduction(

    modelid: int,
    session: Session = Depends(getsession),

):
    """Promosi model ke production"""
    model = session.get(MLModelDB, modelid)
    if not model:
        raise HTTPException(statuscode=404, detail="Model not found")


    # Demote model production yang lama (jika ada)
    currentprod = session.exec(
        select(MLModelDB).where(
            MLModelDB.name == model.name,
            MLModelDB.isproduction == True,

        )
    ).all()
    for m in currentprod:
        m.isproduction = False

        session.add(m)

    model.isproduction = True
    session.add(model)
    session.commit()

    return {"message": f"Model {model.name} v{model.version} promoted to production"}

@app.get("/models/production/{modelname}", responsemodel=MLModelRead)
def getproductionmodel(
    modelname: str,

    session: Session = Depends(getsession),
):
    """Ambil model yang sedang di production"""
    statement = select(MLModelDB).where(
        MLModelDB.name == modelname,

        MLModelDB.isproduction == True,
    )
    model = session.exec(statement).first()
    if not model:
        raise HTTPException(statuscode=404, detail="No production model found")

    return model

@app.get("/models/compare")
def comparemodels(
    modelids: str = Query(description="Comma-separated model IDs"),

    session: Session = Depends(getsession),
):
    """Bandingkan beberapa model"""
    ids = [int(id.strip()) for id in modelids.split(",")]

    models = []
    for modelid in ids:
        model = session.get(MLModelDB, modelid)

        if model:
            models.append({
                "id": model.id,
                "name": model.name,
                "version": model.version,
                "accuracy": model.accuracy,
                "framework": model.framework,
                "isproduction": model.isproduction,

            })
    return {"models": models}

Tips dan Best Practices

Selalu gunakan schema terpisah: Buat Base, Create, Read, dan Update schema untuk setiap model.

Gunakan Field dengan validasi: Manfaatkan parameter ge, le, minlength, maxlength untuk validasi di level model.

Index kolom yang sering di-query: Gunakan index=True pada field yang sering digunakan dalam WHERE clause.

Session management yang benar: Selalu gunakan context manager (with Session) untuk menghindari connection leak.

Migration wajib di production: Jangan gunakan createall() di production. Selalu gunakan Alembic untuk migration.

Async untuk high-throughput: Gunakan async session untuk API yang membutuhkan throughput tinggi.

Kesimpulan

SQLModel adalah ORM modern yang sempurna untuk aplikasi AI/ML Python. Dengan menggabungkan kekuatan SQLAlchemy dan Pydantic, SQLModel memberikan pengalaman development yang type-safe, tervalidasi, dan terintegrasi sempurna dengan FastAPI.

Untuk proyek AI, SQLModel sangat cocok digunakan sebagai backend untuk experiment tracking, model registry, dataset management, dan berbagai kebutuhan data persistence lainnya. Kemudahan penggunaannya tidak mengorbankan kekuatan, karena akses penuh ke SQLAlchemy tetap tersedia saat dibutuhkan.

Mulailah dengan proyek kecil menggunakan SQLite, lalu migrasi ke PostgreSQL saat siap ke production. Dengan Alembic untuk migration dan FastAPI untuk API layer, Anda memiliki stack yang solid untuk membangun platform AI yang scalable.