FastAPI #

FastAPI is a modern web framework for building APIs with Python 3.8+ that combines high performance (on par with Node.js and Go), automatic data validation via Pydantic, and auto-generated interactive documentation. FastAPI is built on Starlette (for web) and Pydantic (for data), supports async/await natively, and follows the OpenAPI and JSON Schema standards. It’s the top choice for microservices, REST APIs, and ML serving in modern production environments.

Installation #

pip install fastapi uvicorn[standard] pydantic-settings

Run the development server:

uvicorn main:app --reload --host 0.0.0.0 --port 8000
# Automatic documentation available at:
# http://localhost:8000/docs      (Swagger UI)
# http://localhost:8000/redoc     (ReDoc)

First Application #

# main.py

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup code (database connection, ML model loading, etc.)
    print("Application started")
    yield
    # Shutdown code (close connections, cleanup, etc.)
    print("Application stopped")

app = FastAPI(
    title="MyApp API",
    description="REST API for my application",
    version="1.0.0",
    lifespan=lifespan
)

@app.get("/")
def root():
    return {"message": "Welcome to MyApp API"}

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

Pydantic Models — Data Validation #

Pydantic is FastAPI’s foundation for validating request bodies, response schemas, and configuration. Defining good models is the key to a reliable API.

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("Product name can't be only numbers")
        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 -- allow from ORM objects

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

Path, Query, and 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="Product ID"),
):
    produk = db_ambil_produk(produk_id)
    if not produk:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {produk_id} not found"
        )
    return produk

@app.get("/produk", response_model=PaginatedResponse)
def daftar_produk(
    q:           Optional[str] = Query(None, min_length=2, description="Search keyword"),
    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),
):
    # Filter and pagination logic
    offset = (halaman - 1) * per_halaman
    # ... database query
    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 is already validated automatically by 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="Product not found")

    # Update only the sent fields (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="Product not found")

Dependency Injection #

Dependency Injection (DI) in FastAPI lets you separate reusable logic — database connections, authentication, pagination — from route handlers.

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 for product filters
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

# Use in a 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}

JWT Authentication #

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", "change-in-production")
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="Invalid token")
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            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="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"}
        )

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

# Use in routes requiring authentication
@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="User not found")
    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 -- allow access from the frontend
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# GZIP compression for large responses
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Custom middleware for request logging
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 are useful for operations that don’t need to finish before a response is sent — like sending emails, updating logs, or processing images.

from fastapi              import BackgroundTasks
from fastapi_mail         import FastMail, MessageSchema

def kirim_email_konfirmasi(email: str, nama: str, order_id: int):
    """This function runs in the background after the response is sent."""
    # Simulate sending an email
    print(f"Sending email to {email} for order #{order_id}")
    # ... real email sending logic

def perbarui_statistik(produk_id: int):
    """Update the view counter in the background."""
    # ... update the 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)
):
    # Create the order in the database
    order_baru = db_buat_order(db, order, int(payload["sub"]))

    # Register background tasks -- run after the response is sent
    background_tasks.add_task(
        kirim_email_konfirmasi,
        email=payload["email"],
        nama=payload.get("nama", ""),
        order_id=order_baru.id
    )

    return order_baru  # the response is sent immediately without waiting for the 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="Product not found")

    background_tasks.add_task(perbarui_statistik, produk_id)
    return produk

Global Error Handling #

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

# Handler for Pydantic validation errors (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": "Invalid data", "errors": errors}
    )

# Handler for database integrity errors (409 Conflict)
@app.exception_handler(IntegrityError)
async def integrity_error_handler(request: Request, exc: IntegrityError):
    return JSONResponse(
        status_code=409,
        content={"detail": "Data already exists or violates a database constraint"}
    )

# Handler for all unhandled exceptions (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": "An internal server error occurred"}
    )

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

# In-memory database for 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 the database dependency for 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

Summary #

Pydantic for all validation — define separate schemas for Create, Update, and Response; use field_validator for custom validation and Field(...) for constraints.
model_dump(exclude_unset=True) — use it for partial updates (PATCH) so only the sent fields are updated, not all fields to their defaults.
Dependency Injection — separate DB connections, authentication, and query parameters into dependency functions/classes used with Depends(); this makes code modular and easy to test.
lifespan not @app.on_event — the modern way to manage startup/shutdown in FastAPI; on_event is deprecated.
Explicit status_code — always include status_code in route decorators (201 for create, 204 for delete); don’t rely on the default 200.
Background tasks for non-blocking operations — use BackgroundTasks for emails, notifications, and stat updates that don’t need to be awaited before the response is sent.
CORS middleware — always configure CORSMiddleware if the frontend is on a different domain; don’t use allow_origins=["*"] in production.
Global error handlers — register handlers for RequestValidationError (422) and general exceptions so error responses are consistent across the whole API.
TestClient + dependency overrides — use app.dependency_overrides to swap the database connection for an in-memory database during testing.
Automatic documentation — FastAPI generates Swagger UI (/docs) and ReDoc (/redoc) from type hints and Pydantic models; make sure description and examples in Field are well filled for informative docs.

← Previous: Django Next: Flask →