Belajar FastAPI - Pagination, Filtering & Versioning
Membangun endpoint list production-grade: limit/offset dan cursor pagination dengan SQLAlchemy, filtering dinamis yang aman, response pagination yang terstandar, serta strategi API versioning via path dan header.
#Pendahuluan
Setelah di episode 19 API kalian terlindungi dari banjir request, sekarang kita bahas pola yang menentukan pengalaman klien saat data berjumlah besar: pagination, filtering, dan versioning. Endpoint GET /items/ yang mengembalikan 10.000 baris bukan hanya lambat — ia juga tidak bisa dipakai.
Mengapa episode ini penting? Pagination, filter, dan versioning adalah fitur yang "tidak terlihat" sampai API dipakai skala nyata — dan justru saat itulah kualitasnya teruji. Episode ini membangun endpoint list yang benar dari awal: cepat, terstandar, dan tidak memecah klien yang sudah ada.
#Limit/Offset Pagination
Model paling sederhana dan paling umum:
from typing import Annotated
from fastapi import Depends, Query
from sqlalchemy import func, select
from app import models, schemas
from app.database import get_db
from app.schemas import ItemOut
@app.get("/items/")
def list_items(
db: Session = Depends(get_db),
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
) -> list[ItemOut]:
statement = (
select(models.Item)
.order_by(models.Item.id)
.offset(skip)
.limit(limit)
)
return db.execute(statement).scalars().all()offset(skip).limit(limit) adalah SQL LIMIT/OFFSET: ambil limit baris mulai dari baris ke-skip. Sederhana dan cukup untuk dataset hingga puluhan ribu baris.
Warning
Limit/offset melambat saat skip besar — database harus memindai dan membuang skip baris dulu (offset 1 juta pada tabel besar = query puluhan detik). Untuk dataset sangat besar atau feed yang berubah cepat, pakai cursor pagination di bawah.
#Cursor Pagination
Cursor pagination memakai kolom berurutan (biasanya id) sebagai penanda: "ambil data setelah ID ini". Stabil saat data ditambah/dihapus, dan konstan terhadap kedalaman:
@app.get("/items/cursor/")
def list_items_cursor(
db: Session = Depends(get_db),
cursor: int | None = Query(None, ge=1),
limit: int = Query(20, ge=1, le=100),
) -> dict:
statement = select(models.Item)
if cursor is not None:
statement = statement.where(models.Item.id > cursor)
statement = statement.order_by(models.Item.id).limit(limit)
items = db.execute(statement).scalars().all()
next_cursor = items[-1].id if len(items) == limit else None
return {
"items": items,
"next_cursor": next_cursor,
}next_cursor adalah id item terakhir halaman ini — klien mengirimnya sebagai cursor di halaman berikutnya. None berarti tidak ada halaman lagi. Keunggulan: tambah/hapus data di tengah tidak membuat halaman dobel/terlewat, dan query selalu cepat.
Tip
Aturan praktis: limit/offset untuk dashboard & data yang bisa di-sort sewenang-wenang; cursor untuk feed, inbox, dan dataset yang bergerak cepat. Beberapa API besar bahkan hanya menyediakan cursor.
#Filtering Dinamis yang Aman
Filter biasanya datang sebagai query param. Kuncinya: selalu petakan nama param ke kolom, jangan pernah menerima nama kolom mentah dari klien:
from sqlalchemy.sql.operators import gt, lt
FILTER_MAP = {
"price_gt": (models.Item.price, gt),
"price_lt": (models.Item.price, lt),
"name": (models.Item.name, lambda col, v: col.ilike(f"%{v}%")),
}
@app.get("/items/filter/")
def list_items_filtered(
db: Session = Depends(get_db),
price_gt: float | None = None,
price_lt: float | None = None,
name: str | None = None,
) -> list[ItemOut]:
statement = select(models.Item)
filters = {
"price_gt": price_gt,
"price_lt": price_lt,
"name": name,
}
for key, value in filters.items():
if value is None:
continue
column, operator = FILTER_MAP[key]
statement = statement.where(operator(column, value))
return db.execute(statement).scalars().all()FILTER_MAP adalah allow-list: hanya filter yang terdaftar yang bisa dipakai. Klien tidak pernah bisa menyuntikkan nama kolom atau operator SQL — nilai filter tetap dikirim sebagai parameter terikat (aman dari SQLi, episode 18).
Important
Jangan pernah membangun filter dari string f"{field}={value}" atau menerima nama kolom dari klien. Peta eksplisit (FILTER_MAP) membuat permukaan serangan kecil dan filter tetap type-safe.
#Response Pagination Terstandar
Klien lebih senang pagination yang konsisten di semua endpoint. Standarkan bentuk response-nya:
from pydantic import BaseModel
class Page(BaseModel):
items: list[ItemOut]
total: int
skip: int
limit: int
next_cursor: int | None = None
@app.get("/items/page/", response_model=Page)
def list_items_page(
db: Session = Depends(get_db),
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
) -> Page:
total = db.scalar(select(func.count()).select_from(models.Item))
items = (
db.execute(
select(models.Item).offset(skip).limit(limit)
).scalars().all()
)
return Page(
items=items,
total=total or 0,
skip=skip,
limit=limit,
)total (dari func.count()) memberi klien pengetahuan untuk render pager "halaman 2 dari 15". Dengan response_model, seluruh bentuk ini terdokumentasi di OpenAPI.
#API Versioning
Kapan API perlu versi baru? Ketika ada perubahan yang memecah klien — misalnya mengubah bentuk response. Dua strategi dominan:
1. Path versioning — paling eksplisit:
app_v1 = FastAPI()
app_v2 = FastAPI()
@api_v1.get("/items/{item_id}")
def get_item_v1(item_id: int) -> dict:
return {"item_id": item_id, "name": "lama"}
@api_v2.get("/items/{item_id}")
def get_item_v2(item_id: int) -> dict:
return {"item_id": item_id, "name": "baru", "tags": []}Rute GET /v1/items/1 dan GET /v2/items/1 (lewat mount atau prefix router). Klien lama tetap berjalan di v1, klien baru pindah ke v2.
2. Header versioning — memakai header kustom seperti Accept-Version: v2:
from fastapi import Header
@app.get("/items/{item_id}")
def get_item(
item_id: int,
accept_version: str | None = Header(default="v1"),
) -> dict:
if accept_version == "v2":
return {"item_id": item_id, "name": "baru", "tags": []}
return {"item_id": item_id, "name": "lama"}Path versioning lebih disukai untuk API publik karena URL tidak ambigu dan bisa di-cache per-version. Header versioning cocok untuk API internal tempat satu klien mengontrol kedua sisi.
Note
Versioning bukan "versi library" — versikan kontrak, bukan kode. Bila perubahan additive (menambah field, endpoint baru) tidak perlu versi baru; versi baru hanya untuk perubahan yang memecah klien. Kunci API jangka panjang ada pada disiplin ini.
#Common Pitfalls
| Pitfall | Solusi |
|---|---|
| Offset besar lambat | Cursor pagination untuk dataset besar |
| Filter dengan nama kolom dari klien | Allow-list FILTER_MAP |
| Response pagination beda-beda bentuk | Schema Page terstandar |
| Versi baru memecah klien lama | Versioning via path/header, pertahankan v1 |
total dihitung per query mahal | Cache atau hitung hanya saat diperlukan |
#Penutup
Inti yang harus dibawa pulang:
- Limit/offset untuk dashboard; cursor untuk feed & dataset besar.
- Filter harus lewat allow-list, bukan nama kolom dari klien.
- Standarkan bentuk page:
items,total,next_cursor. - Versioning untuk perubahan yang memecah klien: path (
/v2) atau header (Accept-Version).
Di episode 21 selanjutnya kita akan membahas OpenAPI & dokumentasi lanjutan — mengkustomisasi metadata, tags, deskripsi endpoint, response docs, hingga schema di auto-docs. Dokumentasi kalian akan jadi aset, bukan hiasan!