Tujuan Pembelajaran

• Memahami konsep dasar FastAPI dan mengapa populer
• Mampu membuat REST API dengan Python
• Mengerti request/response models dengan Pydantic
• Menerapkan authentication dan authorization
• Menggunakan database dengan SQLAlchemy
• Deployment ke production

Keunggulan FastAPI

• Fast - Sangat cepat
• Auto Documentation - Swagger UI otomatis
• Type Safety - Full type hints dengan Pydantic
• Async - Native async/await support

Dasar-Dasar FastAPI

Instalasi & Hello World

# Install FastAPI
pip install fastapi
pip install uvicorn[standard]

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

# Jalankan server
uvicorn main:app --reload

Path Parameters

# Basic Path Parameter
@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

# Enum Path Parameter
from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"

@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
    return {"model_name": model_name}

Query Parameters

# Basic Query
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# URL: /items/?skip=0&limit=10

# Optional Parameters
from typing import Optional

@app.get("/users/")
def read_users(q: Optional[str] = None):
    if q:
        return {"q": q}
    return {"q": "No query"}

Request Body (POST)

from pydantic import BaseModel
from typing import Optional

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

@app.post("/items/")
def create_item(item: Item):
    return item

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}

Response Models

# Filter return fields
class UserIn(BaseModel):
    username: str
    password: str
    email: str

class UserOut(BaseModel):
    username: str
    email: str

@app.post("/users/", response_model=UserOut)
def create_user(user: UserIn):
    # Password tidak akan di-return!
    return user

# Status Codes
from fastapi import status

@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(item: Item):
    return item

Intermediate FastAPI

Error Handling

from fastapi import HTTPException

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id == 0:
        raise HTTPException(status_code=400, detail="Invalid ID")
    return {"item_id": item_id}

# Custom Error Response
@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = find_user(user_id)
    if not user:
        raise HTTPException(
            status_code=404,
            detail={"error": "User not found", "user_id": user_id}
        )
    return user

Dependencies

from fastapi import Depends

# Basic Depends
def get_query_param(q: Optional[str] = None):
    return q

@app.get("/items/")
def read_items(q: Optional[str] = Depends(get_query_param)):
    return {"q": q}

# Database Dependency
def get_db():
    db = connect_to_database()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/")
def get_users(db = Depends(get_db)):
    return db.query(User).all()

Database dengan SQLAlchemy

# Install
pip install sqlalchemy databases[sqlite]

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# models.py
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    price = Column(Float)

# Buat tabel
Base.metadata.create_all(bind=engine)

CRUD Operations

# Create
@app.post("/items/", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

# Read
@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return db.query(models.Item).offset(skip).limit(limit).all()

# Update
@app.put("/items/{item_id}")
def update_item(item_id: int, item: schemas.ItemUpdate, db: Session = Depends(get_db)):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    for key, value in item.dict().items():
        setattr(db_item, key, value)
    db.commit()
    return db_item

# Delete
@app.delete("/items/{item_id}")
def delete_item(item_id: int, db: Session = Depends(get_db)):
    db.query(models.Item).filter(models.Item.id == item_id).delete()
    db.commit()
    return {"ok": True}

Middleware

# CORS
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Timing Middleware
from fastapi import Request
import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Advanced FastAPI

Authentication (JWT)

# Install
pip install python-jose[cryptography] passlib[bcrypt]

# Setup
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Create Token
from datetime import datetime, timedelta

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

Protected Routes

# Login
from fastapi.security import OAuth2PasswordRequestForm

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

# Protected Endpoint
async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
    except JWTError:
        raise HTTPException(status_code=401, detail="Could not validate credentials")
    return get_user(username)

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

File Upload

# Upload File
from fastapi import UploadFile, File

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    with open(f"uploads/{file.filename}", "wb") as f:
        f.write(contents)
    return {"filename": file.filename, "size": len(contents)}

# Download File
from fastapi.responses import FileResponse

@app.get("/download/{filename}")
async def download_file(filename: str):
    return FileResponse(
        path=f"uploads/{filename}",
        filename=filename,
        media_type="application/octet-stream"
    )

WebSockets

from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except Exception:
        await websocket.close()

# Client example (JavaScript)
# const ws = new WebSocket("ws://localhost:8000/ws");
# ws.onmessage = (event) => console.log(event.data);

Selamat! 🎉

Next Steps

• Buat project pribadi dengan FastAPI
• Explore library: SQLModel, Tortoise ORM
• Pelajari deployment: Docker, Railway, Render
• Explore testing dengan pytest

Testing FastAPI

Install pytest

pip install pytest pytest-asyncio httpx

TestClient

# test_main.py
from fastapi.testclient import TestClient
from main import app  # Import FastAPI app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

def test_create_item():
    response = client.post("/items/", json={
        "name": "Test Item",
        "price": 10.5
    })
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Test Item"
    assert "id" in data

Testing dengan Dependencies

# Override dependency
from unittest.mock import MagicMock

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

app.dependency_overrides[get_db] = override_get_db

# Run tests
pytest test_main.py -v

Async Testing

# test_async.py
import pytest
from httpx import AsyncClient, ASGITransport
from main import app

@pytest.mark.asyncio
async def test_async():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/")
        assert response.status_code == 200

Background Tasks

BackgroundTasks

from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # Email sending logic
    print(f"Sending email to {email}: {message}")

@app.post("/send-notification/")
async def send_notification(email: str, message: str, background_tasks: BackgroundTasks):
    # Task akan berjalan setelah response dikirim
    background_tasks.add_task(send_email, email, message)
    return {"message": "Notification queued"}

Celery untuk Heavy Tasks

# tasks.py
from celery import Celery

celery_app = Celery(broker="redis://localhost:6379")

@celery_app.task
def send_mass_email(emails: list, message: str):
    for email in emails:
        # Heavy email sending
        pass

Kapan Gunakan?

Celery + FastAPI Example

# API endpoint
from tasks import send_mass_email

@app.post("/mass-email/")
async def trigger_mass_email(emails: list):
    task = send_mass_email.delay(emails, "Hello!")
    return {"task_id": task.id, "status": "queued"}

# Poll status
@app.get("/task-status/{task_id}")
async def get_status(task_id: str):
    task = send_mass_email.AsyncResult(task_id)
    return {"status": task.state, "result": task.result if task.ready() else None}