Praktik: Database dengan SQLModel & Dependency Injection

Kenapa Butuh Database Sungguhan?

API kita di artikel sebelumnya sudah punya CRUD lengkap dengan validasi yang layak, tapi datanya masih hidup di list Python — hilang setiap kali server di-restart, dan tidak bisa dipakai bersama kalau ada lebih dari satu proses server berjalan. Di artikel ini kita sambungkan aplikasi ke database sungguhan (SQLite, lewat SQLModel), sekaligus membahas tuntas konsep dependency injection yang sempat disinggung sekilas di Bagian 2.

SQLModel — Satu Model untuk Validasi dan Database

SQLModel adalah library, dibuat oleh pencipta FastAPI sendiri, yang menggabungkan Pydantic (validasi data) dengan SQLAlchemy (ORM/database). Manfaat utamanya: kita bisa mendefinisikan satu class yang sekaligus jadi skema Pydantic DAN tabel database — tidak perlu menulis dua definisi terpisah yang gampang tidak sinkron.

pip install sqlmodel

from typing import Optional
from sqlmodel import Field, SQLModel


class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    done: bool = False
    priority: Priority = Priority.medium

Perhatikan table=True — inilah yang membedakan Task dari model Pydantic biasa: class ini juga merepresentasikan tabel SQL sungguhan. id bertipe Optional[int] dengan default None karena nilainya baru diisi database (auto-increment) setelah row-nya disimpan, bukan oleh kita.

Engine dan Session — Cara SQLModel Bicara dengan Database

Dua konsep dasar sebelum bisa query:

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

DATABASE_URL = "sqlite:///./tasks.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})


def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)


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

connect_args={"check_same_thread": False} khusus dibutuhkan SQLite karena defaultnya SQLite hanya izinkan diakses dari thread yang membuat koneksinya — sementara FastAPI bisa memproses request dari thread berbeda. Untuk database lain (PostgreSQL, MySQL), baris ini tidak diperlukan.

Perhatikan get_session memakai yield, bukan return. Ini polanya generator — kita akan bahas kenapa ini penting di bagian berikutnya.

Dependency Injection — Menyelesaikan Janji dari Bagian 2

Sekarang saatnya membahas Depends() secara utuh. Dependency injection adalah cara FastAPI menyediakan sesuatu yang dibutuhkan path operation function — di kasus kita, sebuah Session database — tanpa endpoint itu sendiri yang harus tahu cara membuatnya.

from fastapi import Depends
from sqlmodel import Session

from .database import get_session


@app.get("/api/tasks")
def list_tasks(session: Session = Depends(get_session)):
    ...

Begitu FastAPI melihat Depends(get_session), dia akan: memanggil get_session(), menjalankannya sampai yield (di titik itu, session diberikan ke fungsi list_tasks), lalu setelah list_tasks selesai (baik sukses maupun error), FastAPI melanjutkan eksekusi get_session setelah yield — yang dalam kasus ini menutup session lewat with Session(engine) as session.

Kenapa ini lebih baik dibanding membuat session manual di setiap endpoint?

Merapikan Struktur Project

Sekarang project kita mulai punya beberapa concern berbeda (model, koneksi database, endpoint). Saatnya pecah dari satu file main.py menjadi package yang lebih terstruktur:

task-tracker-api/
├── app/
│   ├── main.py             # setup FastAPI app, lifespan, health check
│   ├── database.py         # engine, get_session
│   ├── models.py            # Task (SQLModel, table=True), Priority
│   ├── schemas.py           # TaskCreate, TaskUpdate, TaskRead
│   └── routers/
│       └── tasks.py         # semua endpoint /api/tasks/*
└── tests/
    └── test_tasks.py

APIRouter dipakai untuk mengelompokkan endpoint yang berhubungan ke file terpisah dari main.py:

# routers/tasks.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select

from app.database import get_session
from app.models import Task
from app.schemas import TaskCreate, TaskRead

router = APIRouter(prefix="/api/tasks", tags=["tasks"])


@router.post("", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate, session: Session = Depends(get_session)):
    task = Task(title=payload.title, priority=payload.priority)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

prefix="/api/tasks" berarti kita tidak perlu mengulang /api/tasks di setiap decorator — cukup @router.post("") untuk POST /api/tasks, @router.get("/{task_id}") untuk GET /api/tasks/{task_id}, dan seterusnya. Pola session.add() → session.commit() → session.refresh() adalah alur standar SQLModel/SQLAlchemy untuk insert: tambahkan objek ke session, simpan ke database, lalu ambil ulang objeknya (supaya id yang di-generate database ikut terisi).

Untuk query dengan filter, SQLModel memakai select(), mirip SQL biasa tapi tetap type-safe:

@router.get("", response_model=list[TaskRead])
def list_tasks(done: bool | None = None, session: Session = Depends(get_session)):
    query = select(Task)
    if done is not None:
        query = query.where(Task.done == done)
    return session.exec(query).all()

Terakhir, main.py tinggal menggabungkan semuanya, dan memastikan tabel database dibuat saat aplikasi start lewat lifespan handler:

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI

from app.database import create_db_and_tables
from app.routers import tasks


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(title="Task Tracker API", lifespan=lifespan)


@app.get("/health")
def health_check():
    return {"status": "ok"}


app.include_router(tasks.router)

lifespan adalah cara modern FastAPI menjalankan kode saat aplikasi start (sebelum yield) dan saat aplikasi shutdown (setelah yield) — di kasus kita, cukup memastikan tabel sudah ada sebelum menerima request pertama.

Menjalankan dan Menguji

pip install -r requirements.txt
uvicorn app.main:app --reload

Coba buat beberapa task lewat /docs atau curl, lalu restart server-nya. Berbeda dengan sebelumnya, task yang sudah dibuat tetap ada — karena sekarang datanya tersimpan di file tasks.db, bukan di memory proses yang hilang begitu proses berhenti.

Rangkuman