FastAPI #

FastAPI adalah web framework modern untuk membangun API dengan Python 3.8+ yang menggabungkan kecepatan performa tinggi (setara Node.js dan Go), validasi data otomatis via Pydantic, dan dokumentasi interaktif yang di-generate otomatis. FastAPI dibangun di atas Starlette (untuk web) dan Pydantic (untuk data), mendukung async/await secara native, dan mengikuti standar OpenAPI dan JSON Schema. Pilihan utama untuk microservice, REST API, dan ML serving di lingkungan produksi modern.

Instalasi #

pip install fastapi uvicorn[standard] pydantic-settings

Jalankan server development:

uvicorn main:app --reload --host 0.0.0.0 --port 8000
# Dokumentasi otomatis tersedia di:
# http://localhost:8000/docs      (Swagger UI)
# http://localhost:8000/redoc     (ReDoc)

Aplikasi Pertama #

# main.py

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Kode saat startup (koneksi database, load model ML, dll.)
    print("Aplikasi dimulai")
    yield
    # Kode saat shutdown (tutup koneksi, cleanup, dll.)
    print("Aplikasi berhenti")

app = FastAPI(
    title="MyApp API",
    description="REST API untuk aplikasi saya",
    version="1.0.0",
    lifespan=lifespan
)

@app.get("/")
def root():
    return {"message": "Selamat datang di MyApp API"}

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

Pydantic Models — Validasi Data #

Pydantic adalah fondasi FastAPI untuk validasi request body, response schema, dan konfigurasi. Mendefinisikan model yang baik adalah kunci API yang andal.

from pydantic import BaseModel, Field, EmailStr, field_validator
from typing   import Optional, List
from decimal  import Decimal
from datetime import datetime
from enum     import Enum

class StatusProduk(str, Enum):
    aktif    = "aktif"
    nonaktif = "nonaktif"
    habis    = "habis"

class ProdukBase(BaseModel):
    nama:      str     = Field(..., min_length=2, max_length=200, examples=["Laptop Gaming ASUS"])
    deskripsi: str     = Field(default="", max_length=5000)
    harga:     Decimal = Field(..., gt=0, decimal_places=2, examples=[18500000])
    stok:      int     = Field(default=0, ge=0)
    status:    StatusProduk = StatusProduk.aktif

    @field_validator("nama")
    @classmethod
    def nama_tidak_boleh_angka_saja(cls, v):
        if v.strip().isdigit():
            raise ValueError("Nama produk tidak boleh hanya angka")
        return v.strip()

class ProdukCreate(ProdukBase):
    kategori_id: Optional[int] = None

class ProdukUpdate(BaseModel):
    nama:       Optional[str]          = Field(None, min_length=2, max_length=200)
    harga:      Optional[Decimal]      = Field(None, gt=0)
    stok:       Optional[int]          = Field(None, ge=0)
    status:     Optional[StatusProduk] = None
    kategori_id: Optional[int]         = None

class ProdukResponse(ProdukBase):
    id:         int
    slug:       str
    dibuat_pada: datetime
    diubah_pada: datetime

    model_config = {"from_attributes": True}  # Pydantic V2 -- izinkan dari ORM object

class PaginatedResponse(BaseModel):
    total:    int
    halaman:  int
    per_halaman: int
    data:     List[ProdukResponse]

Path, Query, dan Body Parameters #

from fastapi import FastAPI, Path, Query, Body, HTTPException, status
from typing  import Optional, List

app = FastAPI()

@app.get("/produk/{produk_id}", response_model=ProdukResponse)
def ambil_produk(
    produk_id: int = Path(..., gt=0, description="ID produk"),
):
    produk = db_ambil_produk(produk_id)
    if not produk:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Produk dengan ID {produk_id} tidak ditemukan"
        )
    return produk

@app.get("/produk", response_model=PaginatedResponse)
def daftar_produk(
    q:           Optional[str] = Query(None, min_length=2, description="Kata kunci pencarian"),
    kategori_id: Optional[int] = Query(None, gt=0),
    min_harga:   Optional[Decimal] = Query(None, gt=0),
    max_harga:   Optional[Decimal] = Query(None, gt=0),
    status:      Optional[StatusProduk] = Query(None),
    halaman:     int = Query(1, ge=1),
    per_halaman: int = Query(20, ge=1, le=100),
):
    # Logika filter dan pagination
    offset = (halaman - 1) * per_halaman
    # ... query database
    return PaginatedResponse(total=0, halaman=halaman, per_halaman=per_halaman, data=[])

@app.post("/produk", response_model=ProdukResponse, status_code=status.HTTP_201_CREATED)
def buat_produk(produk: ProdukCreate):
    # produk sudah tervalidasi otomatis oleh Pydantic
    hasil = db_buat_produk(produk)
    return hasil

@app.patch("/produk/{produk_id}", response_model=ProdukResponse)
def update_produk(
    produk_id: int = Path(..., gt=0),
    data:      ProdukUpdate = Body(...)
):
    produk = db_ambil_produk(produk_id)
    if not produk:
        raise HTTPException(status_code=404, detail="Produk tidak ditemukan")

    # Update hanya field yang dikirim (exclude_unset=True)
    update_data = data.model_dump(exclude_unset=True)
    return db_update_produk(produk_id, update_data)

@app.delete("/produk/{produk_id}", status_code=status.HTTP_204_NO_CONTENT)
def hapus_produk(produk_id: int = Path(..., gt=0)):
    if not db_hapus_produk(produk_id):
        raise HTTPException(status_code=404, detail="Produk tidak ditemukan")

Dependency Injection #

Dependency Injection (DI) di FastAPI memungkinkan kamu memisahkan logika yang reusable — koneksi database, autentikasi, pagination — dari handler route.

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from typing import Generator, Optional

# Database session dependency
def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Pagination dependency
class PaginationParams:
    def __init__(
        self,
        halaman:     int = Query(1, ge=1),
        per_halaman: int = Query(20, ge=1, le=100)
    ):
        self.halaman     = halaman
        self.per_halaman = per_halaman
        self.offset      = (halaman - 1) * per_halaman

# Dependency untuk filter produk
class ProdukFilter:
    def __init__(
        self,
        q:          Optional[str] = Query(None),
        kategori_id: Optional[int] = Query(None),
        status:     Optional[StatusProduk] = Query(None)
    ):
        self.q           = q
        self.kategori_id = kategori_id
        self.status      = status

# Gunakan di route
@app.get("/produk")
def daftar_produk(
    pagination: PaginationParams  = Depends(PaginationParams),
    filter:     ProdukFilter      = Depends(ProdukFilter),
    db:         Session           = Depends(get_db)
):
    query = db.query(Produk)

    if filter.q:
        query = query.filter(Produk.nama.ilike(f"%{filter.q}%"))
    if filter.kategori_id:
        query = query.filter(Produk.kategori_id == filter.kategori_id)
    if filter.status:
        query = query.filter(Produk.status == filter.status)

    total   = query.count()
    produk  = query.offset(pagination.offset).limit(pagination.per_halaman).all()

    return {"total": total, "data": produk}

Autentikasi JWT #

from fastapi         import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose            import JWTError, jwt
from passlib.context import CryptContext
from datetime        import datetime, timedelta
from pydantic        import BaseModel
import os

SECRET_KEY  = os.getenv("SECRET_KEY", "ganti-di-produksi")
ALGORITHM   = "HS256"
TOKEN_EXPIRE_MENIT = 30

pwd_context       = CryptContext(schemes=["bcrypt"])
oauth2_scheme     = OAuth2PasswordBearer(tokenUrl="/auth/token")

class TokenResponse(BaseModel):
    access_token: str
    token_type:   str = "bearer"

def buat_token(data: dict, expire_menit: int = TOKEN_EXPIRE_MENIT) -> str:
    payload = data.copy()
    payload["exp"] = datetime.utcnow() + timedelta(minutes=expire_menit)
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verifikasi_token(token: str = Depends(oauth2_scheme)) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        if not user_id:
            raise HTTPException(status_code=401, detail="Token tidak valid")
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token tidak valid atau kadaluarsa",
            headers={"WWW-Authenticate": "Bearer"}
        )

@app.post("/auth/token", response_model=TokenResponse)
def login(form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    pengguna = db.query(Pengguna).filter(Pengguna.email == form.username).first()

    if not pengguna or not pwd_context.verify(form.password, pengguna.password_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Email atau password salah",
            headers={"WWW-Authenticate": "Bearer"}
        )

    token = buat_token({"sub": str(pengguna.id), "email": pengguna.email})
    return TokenResponse(access_token=token)

# Gunakan di route yang butuh autentikasi
@app.get("/profil/saya")
def profil_saya(
    payload: dict    = Depends(verifikasi_token),
    db:      Session = Depends(get_db)
):
    pengguna = db.query(Pengguna).filter(Pengguna.id == int(payload["sub"])).first()
    if not pengguna:
        raise HTTPException(status_code=404, detail="Pengguna tidak ditemukan")
    return pengguna

Middleware #

from fastapi           import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logging

logger = logging.getLogger(__name__)

# CORS -- izinkan akses dari frontend
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Kompresi GZIP untuk response besar
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Custom middleware untuk logging request
class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        mulai    = time.time()
        response = await call_next(request)
        durasi   = (time.time() - mulai) * 1000

        logger.info(
            f"{request.method} {request.url.path} "
            f"→ {response.status_code} ({durasi:.1f}ms)"
        )
        return response

app.add_middleware(RequestLoggingMiddleware)

Background Tasks #

Background tasks berguna untuk operasi yang tidak perlu menunggu selesai sebelum response dikirim — seperti mengirim email, memperbarui log, atau memproses gambar.

from fastapi              import BackgroundTasks
from fastapi_mail         import FastMail, MessageSchema

def kirim_email_konfirmasi(email: str, nama: str, order_id: int):
    """Fungsi ini dijalankan di background setelah response dikirim."""
    # Simulasi kirim email
    print(f"Mengirim email ke {email} untuk order #{order_id}")
    # ... logika pengiriman email sesungguhnya

def perbarui_statistik(produk_id: int):
    """Update view counter di background."""
    # ... update database

@app.post("/orders", response_model=OrderResponse, status_code=201)
def buat_order(
    order:           OrderCreate,
    background_tasks: BackgroundTasks,
    db:              Session = Depends(get_db),
    payload:         dict    = Depends(verifikasi_token)
):
    # Buat order di database
    order_baru = db_buat_order(db, order, int(payload["sub"]))

    # Daftarkan background tasks -- dijalankan setelah response dikirim
    background_tasks.add_task(
        kirim_email_konfirmasi,
        email=payload["email"],
        nama=payload.get("nama", ""),
        order_id=order_baru.id
    )

    return order_baru  # response dikirim segera tanpa menunggu email

@app.get("/produk/{produk_id}")
def detail_produk(
    produk_id:        int           = Path(..., gt=0),
    background_tasks: BackgroundTasks = BackgroundTasks(),
    db:               Session       = Depends(get_db)
):
    produk = db_ambil_produk(db, produk_id)
    if not produk:
        raise HTTPException(status_code=404, detail="Produk tidak ditemukan")

    background_tasks.add_task(perbarui_statistik, produk_id)
    return produk

Error Handling Global #

from fastapi              import FastAPI, Request
from fastapi.responses    import JSONResponse
from fastapi.exceptions   import RequestValidationError
from sqlalchemy.exc       import IntegrityError

# Handler untuk validation error Pydantic (422)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field":   " → ".join(str(loc) for loc in error["loc"][1:]),
            "message": error["msg"],
            "type":    error["type"]
        })
    return JSONResponse(
        status_code=422,
        content={"detail": "Data tidak valid", "errors": errors}
    )

# Handler untuk database integrity error (409 Conflict)
@app.exception_handler(IntegrityError)
async def integrity_error_handler(request: Request, exc: IntegrityError):
    return JSONResponse(
        status_code=409,
        content={"detail": "Data sudah ada atau melanggar constraint database"}
    )

# Handler untuk semua exception yang tidak tertangani (500)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Terjadi kesalahan internal server"}
    )

Testing #

# tests/test_produk.py

import pytest
from fastapi.testclient import TestClient
from sqlalchemy         import create_engine
from sqlalchemy.orm     import sessionmaker
from main               import app, get_db
from database           import Base

# Database in-memory untuk testing
SQLALCHEMY_TEST_URL = "sqlite:///./test.db"
engine_test         = create_engine(SQLALCHEMY_TEST_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(bind=engine_test)

def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

# Override dependency database untuk testing
app.dependency_overrides[get_db] = override_get_db

@pytest.fixture(autouse=True)
def setup_db():
    Base.metadata.create_all(bind=engine_test)
    yield
    Base.metadata.drop_all(bind=engine_test)

client = TestClient(app)

def test_buat_produk():
    response = client.post("/produk", json={
        "nama":  "Laptop Test",
        "harga": 15000000,
        "stok":  5
    })
    assert response.status_code == 201
    data = response.json()
    assert data["nama"]  == "Laptop Test"
    assert data["harga"] == "15000000.00"
    assert "id" in data

def test_buat_produk_harga_negatif():
    response = client.post("/produk", json={
        "nama":  "Laptop Invalid",
        "harga": -1000,
    })
    assert response.status_code == 422
    errors = response.json()["errors"]
    assert any("harga" in e["field"] for e in errors)

def test_ambil_produk_tidak_ada():
    response = client.get("/produk/9999")
    assert response.status_code == 404

Ringkasan #

Pydantic untuk semua validasi — definisikan schema terpisah untuk Create, Update, dan Response; gunakan field_validator untuk validasi kustom dan Field(...) untuk constraint.
model_dump(exclude_unset=True) — gunakan saat partial update (PATCH) agar hanya field yang dikirim yang diupdate, bukan semua field ke nilai default.
Dependency Injection — pisahkan koneksi DB, autentikasi, dan parameter query ke fungsi/kelas dependency yang di-Depends(); ini membuat kode modular dan mudah di-test.
lifespan bukan @app.on_event — cara modern mengelola startup/shutdown di FastAPI; on_event sudah deprecated.
status_code eksplisit — selalu cantumkan status_code di decorator route (201 untuk create, 204 untuk delete); jangan andalkan default 200.
Background tasks untuk operasi non-blocking — gunakan BackgroundTasks untuk email, notifikasi, dan update statistik yang tidak perlu ditunggu sebelum response dikirim.
CORS middleware — selalu konfigurasi CORSMiddleware jika frontend berada di domain berbeda; jangan gunakan allow_origins=["*"] di produksi.
Error handler global — daftarkan handler untuk RequestValidationError (422) dan exception umum agar respons error konsisten di seluruh API.
TestClient + dependency override — gunakan app.dependency_overrides untuk mengganti koneksi database dengan database in-memory saat testing.
Dokumentasi otomatis — FastAPI generate Swagger UI (/docs) dan ReDoc (/redoc) dari type hints dan Pydantic models; pastikan description dan examples di Field terisi dengan baik untuk dokumentasi yang informatif.

← Sebelumnya: Django Berikutnya: Flask →