Belajar FastAPI - CRUD dengan SQL (SQLAlchemy)

#Pendahuluan

Setelah delapan episode membangun fondasi — routing, Pydantic, body, response, dan dependency — sekarang kita rakit semuanya menjadi satu aplikasi yang benar-benar menyimpan data: CRUD penuh dengan SQLAlchemy. Inilah momen di mana FastAPI berubah dari "framework HTTP" menjadi "backend sungguhan".

Mengapa episode ini penting? Karena pola yang kita bangun di sini — ORM model, sesi sebagai dependency, schema input/output terpisah — akan diulang di hampir semua proyek FastAPI di dunia nyata. Kuasai pola ini sekali, dan kalian bisa membangun aplikasi apa pun di atasnya.

#Arsitektur: ORM Model vs Pydantic Schema

Hal pertama yang membingungkan pemula FastAPI: ada dua dunia objek yang mirip tapi berbeda.

Mereka tidak boleh dicampur: endpoint menerima/ mengembalikan Pydantic schema, sementara ORM model hanya hidup di layer database. Jembatan antar keduanya adalah fungsi konversi.

#Setup Database dan Model

app/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
 
DATABASE_URL = "sqlite:///./app.db"
 
engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
)
SessionLocal = sessionmaker(bind=engine, autoflush=False)
 
class Base(DeclarativeBase):
    pass
 
 
def get_db():
    db: Session = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Perhatikan dua hal:

• connect_args={"check_same_thread": False} — khusus SQLite agar bisa dipakai dari thread pool FastAPI; dihapus untuk PostgreSQL.
• get_db() memakai pola dependency yield dari episode 8: sesi dibuat per-request dan selalu ditutup lewat finally.

app/models.py

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base
 
 
class Item(Base):
    __tablename__ = "items"
 
    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(100))
    price: Mapped[float]

Ini SQLAlchemy 2.0 style: tipe dideklarasikan dengan Mapped[...] dan mapped_column, bukan Column lama. Hasilnya — type hints yang bisa dicek mypy.

#Schema Input/Output

app/schemas.py

from pydantic import BaseModel, ConfigDict, Field
 
 
class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)
 
 
class ItemOut(ItemCreate):
    id: int
    model_config = ConfigDict(from_attributes=True)

model_config = ConfigDict(from_attributes=True) adalah kunci integrasi: ini mengizinkan Pydantic membaca atribut dari objek ORM langsung — jadi kita bisa melempar objek Item SQLAlchemy ke response tanpa konversi manual.

#CRUD Endpoint

app/main.py

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from typing import Annotated
 
from app import models, schemas
from app.database import get_db
 
app = FastAPI()
 
DB = Annotated[Session, Depends(get_db)]
 
 
@app.post("/items/", response_model=schemas.ItemOut, status_code=201)
def create_item(item: schemas.ItemCreate, db: DB) -> models.Item:
    db_item = models.Item(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item
 
 
@app.get("/items/{item_id}", response_model=schemas.ItemOut)
def read_item(item_id: int, db: DB) -> models.Item:
    item = db.get(models.Item, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item tidak ditemukan")
    return item
 
 
@app.delete("/items/{item_id}", status_code=204)
def delete_item(item_id: int, db: DB) -> None:
    item = db.get(models.Item, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item tidak ditemukan")
    db.delete(item)
    db.commit()

Alur tiap endpoint konsisten: validasi via Pydantic di pintu masuk → operasi ORM dengan sesi dari dependency → response dikendalikan response_model. Pola DB = Annotated[Session, Depends(get_db)] membuat tiap handler hanya perlu db: DB.

#Migrasi dengan Alembic

Seiring waktu, skema database pasti berubah. Alembic adalah tool migrasi SQLAlchemy resmi:

pip install alembic
alembic init alembic

Konfigurasi alembic/env.py agar memakai metadata model kalian:

alembic/env.py (potongan)

from app.database import Base
from app import models  # import agar model terdaftar
 
target_metadata = Base.metadata

Lalu jalankan migrasi:

alembic revision --autogenerate -m "create items table"
alembic upgrade head

--autogenerate membandingkan skema database dengan model Python dan membuat file migrasi — kalian tinggal review diff-nya sebelum dijalankan.

#PostgreSQL untuk Produksi

Untuk produksi, ganti DATABASE_URL dan install driver:

pip install "psycopg[binary]"

app/database.py - PostgreSQL

DATABASE_URL = "postgresql+psycopg://user:password@localhost:5432/appdb"
 
engine = create_engine(DATABASE_URL, pool_pre_ping=True)

pool_pre_ping=True mencegah "stale connection" saat database restart — setiap peminjaman koneksi dari pool divalidasi dulu.

#Common Pitfalls

#Penutup

Inti yang harus dibawa pulang:

• Dua dunia objek: SQLAlchemy model (DB) dan Pydantic schema (API) — jangan dicampur.
• get_db() sebagai dependency yield mengelola sesi per-request.
• from_attributes=True menghubungkan ORM object ke Pydantic output.
• commit + refresh wajib setelah add.
• Alembic untuk migrasi skema; PostgreSQL untuk produksi.

Di episode 10 selanjutnya kita akan menulis testing untuk aplikasi ini — memakai TestClient (httpx), pytest fixtures, dan dependency override yang membuat test tidak menyentuh database sungguhan. Ini senjata yang mengubah aplikasi dari "jalan di laptop" menjadi "bisa dideploy dengan percaya diri"!