FastAPI: Membuat REST API Modern dengan Python

# Buat direktori proyek
mkdir fastapi-project && cd fastapi-project

# Buat virtual environment
python3 -m venv venv

# Aktifkan virtual environment
# Linux / macOS:
source venv/bin/activate

# Windows (PowerShell):
# .\venv\Scripts\Activate.ps1

# Windows (CMD):
# .\venv\Scripts\activate.bat

# Instalasi semua dependencies sekaligus
pip install fastapi "uvicorn[standard]" \
  "python-jose[cryptography]" "passlib[bcrypt]" \
  python-multipart aiosqlite

# Simpan ke requirements.txt
pip freeze > requirements.txt

# Untuk reinstal di perangkat lain:
# pip install -r requirements.txt

fastapi-project/
├── venv/
├── main.py              ← Entry point aplikasi
├── database.py          ← Konfigurasi database SQLite
├── models.py            ← Model SQLAlchemy/Pydantic
├── schemas.py           ← Schema validasi Pydantic
├── auth.py              ← Logika autentikasi JWT
├── routers/
│   ├── __init__.py
│   ├── items.py         ← Endpoint untuk items
│   └── users.py         ← Endpoint untuk users
└── requirements.txt

"""
Hello World API dengan FastAPI
BeebaneLabs - https://beebanelabs.pages.dev
"""

from fastapi import FastAPI

# Membuat instance aplikasi FastAPI
app = FastAPI(
    title="BeebaneLabs API",
    description="Tutorial REST API dengan FastAPI",
    version="1.0.0"
)


# Endpoint root — GET /
@app.get("/")
async def root():
    """Endpoint sederhana untuk mengecek apakah API berjalan."""
    return {"message": "Hello World!", "status": "running"}


# Endpoint health check — GET /health
@app.get("/health")
async def health_check():
    """Endpoint health check untuk monitoring."""
    return {"status": "healthy", "service": "fastapi-tutorial"}


# Jalankan dengan:
# uvicorn main:app --reload

# Jalankan dengan reload otomatis saat kode berubah
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# Output:
# INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
# INFO:     Started reloader process
# INFO:     Started server process
# INFO:     Waiting for application startup.
# INFO:     Application startup complete.

from fastapi import FastAPI, HTTPException

app = FastAPI()

# Simpan data di memory (sementara)
items_db = {
    1: {"id": 1, "name": "Laptop ASUS", "price": 12000000},
    2: {"id": 2, "name": "Keyboard Mekanik", "price": 850000},
    3: {"id": 3, "name": "Monitor 27 inch", "price": 4500000},
}


@app.get("/items/{item_id}")
async def get_item(item_id: int):
    """Mengambil item berdasarkan ID (path parameter)."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item tidak ditemukan")
    return items_db[item_id]


# Contoh: GET /items/1
# Output: {"id": 1, "name": "Laptop ASUS", "price": 12000000}

# Contoh: GET /items/99
# Output: HTTP 404 — {"detail": "Item tidak ditemukan"}

from typing import Optional

@app.get("/items/")
async def list_items(
    skip: int = 0,              # Default: 0
    limit: int = 10,            # Default: 10
    search: Optional[str] = None,  # Opsional
    sort_by: str = "name"       # Default: "name"
):
    """
    Mengambil daftar item dengan pagination dan filter.
    Contoh: GET /items/?skip=0&limit=5&search=laptop&sort_by=price
    """
    results = list(items_db.values())

    # Filter berdasarkan search
    if search:
        results = [
            item for item in results
            if search.lower() in item["name"].lower()
        ]

    # Sorting
    if sort_by == "price":
        results.sort(key=lambda x: x["price"])
    else:
        results.sort(key=lambda x: x["name"])

    # Pagination
    total = len(results)
    results = results[skip : skip + limit]

    return {
        "total": total,
        "skip": skip,
        "limit": limit,
        "items": results
    }

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime


class ItemCreate(BaseModel):
    """Schema untuk membuat item baru (request body)."""
    name: str = Field(..., min_length=2, max_length=100, 
                      contoh="Laptop ASUS ROG")
    price: int = Field(..., gt=0, contoh=15000000)
    description: Optional[str] = Field(None, max_length=500)
    is_available: bool = True

    class Config:
        # Contoh untuk dokumentasi Swagger
        json_schema_extra = {
            "contoh": {
                "name": "Laptop ASUS ROG",
                "price": 15000000,
                "description": "Laptop gaming high-end",
                "is_available": True
            }
        }


class ItemUpdate(BaseModel):
    """Schema untuk update item (semua field opsional)."""
    name: Optional[str] = Field(None, min_length=2, max_length=100)
    price: Optional[int] = Field(None, gt=0)
    description: Optional[str] = None
    is_available: Optional[bool] = None


class ItemResponse(BaseModel):
    """Schema untuk response (output) item."""
    id: int
    name: str
    price: int
    description: Optional[str] = None
    is_available: bool
    created_at: str

from fastapi import FastAPI, HTTPException, status

app = FastAPI()
next_id = 4  # ID auto-increment sederhana

# Database sementara (di memory)
items_db = {
    1: {"id": 1, "name": "Laptop ASUS", "price": 12000000,
        "description": None, "is_available": True,
        "created_at": "2026-06-25T10:00:00"},
    2: {"id": 2, "name": "Keyboard Mekanik", "price": 850000,
        "description": "Cherry MX Red", "is_available": True,
        "created_at": "2026-06-25T10:05:00"},
}


# CREATE — POST /items
@app.post("/items/", response_model=ItemResponse, 
          status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemCreate):
    """Membuat item baru. FastAPI otomatis memvalidasi body."""
    global next_id

    new_item = {
        "id": next_id,
        "name": item.name,
        "price": item.price,
        "description": item.description,
        "is_available": item.is_available,
        "created_at": datetime.now().isoformat()
    }
    items_db[next_id] = new_item
    next_id += 1

    return new_item


# UPDATE — PUT /items/{item_id}
@app.put("/items/{item_id}", response_model=ItemResponse)
async def update_item(item_id: int, item: ItemUpdate):
    """Update item. Hanya field yang dikirim yang berubah."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item tidak ditemukan")

    existing = items_db[item_id]

    # Update hanya field yang tidak None (partial update)
    update_data = item.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        existing[key] = value

    return existing


# DELETE — DELETE /items/{item_id}
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    """Menghapus item berdasarkan ID."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item tidak ditemukan")
    del items_db[item_id]
    return None

"""
Konfigurasi Database SQLAlchemy + SQLite
BeebaneLabs - https://beebanelabs.pages.dev
"""

from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

# SQLite database URL — file.db akan dibuat otomatis
DATABASE_URL = "sqlite:///./app.db"

# Buat engine SQLAlchemy
engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # Diperlukan untuk SQLite
)

# Buat session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base class untuk model ORM
Base = declarative_base()


# ===== MODEL DATABASE =====
class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    name = Column(String(100), nullable=False, index=True)
    price = Column(Integer, nullable=False)
    description = Column(String(500), nullable=True)
    is_available = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.now)
    owner_id = Column(Integer, nullable=True)


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.now)


def init_db():
    """Membuat semua tabel di database jika belum ada."""
    Base.metadata.create_all(bind=engine)


def get_db():
    """Dependency injection untuk mendapatkan database session."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from database import init_db, get_db, Item, User
from schemas import ItemCreate, ItemResponse, ItemUpdate

app = FastAPI(title="BeebaneLabs API", version="1.0.0")


@app.on_event("startup")
async def startup():
    """Inisialisasi database saat server pertama kali start."""
    init_db()
    print("[OK] Database terinisialisasi")


# CREATE — POST /items/
@app.post("/items/", response_model=ItemResponse,
          status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    """Membuat item baru dan menyimpan ke database SQLite."""
    db_item = Item(
        name=item.name,
        price=item.price,
        description=item.description,
        is_available=item.is_available
    )
    db.add(db_item)        # Tambahkan ke session
    db.commit()            # Simpan ke database
    db.refresh(db_item)    # Refresh untuk mendapatkan ID dan default values
    return db_item


# READ ALL — GET /items/
@app.get("/items/", response_model=list[ItemResponse])
async def list_items(
    skip: int = 0,
    limit: int = 10,
    search: str = None,
    db: Session = Depends(get_db)
):
    """Mengambil daftar item dari database dengan pagination."""
    query = db.query(Item)

    if search:
        query = query.filter(Item.name.contains(search))

    items = query.offset(skip).limit(limit).all()
    return items


# READ ONE — GET /items/{item_id}
@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int, db: Session = Depends(get_db)):
    """Mengambil satu item berdasarkan ID."""
    item = db.query(Item).filter(Item.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=404, detail="Item tidak ditemukan")
    return item


# UPDATE — PUT /items/{item_id}
@app.put("/items/{item_id}", response_model=ItemResponse)
async def update_item(
    item_id: int,
    item_update: ItemUpdate,
    db: Session = Depends(get_db)
):
    """Update item di database."""
    item = db.query(Item).filter(Item.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=404, detail="Item tidak ditemukan")

    update_data = item_update.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(item, key, value)

    db.commit()
    db.refresh(item)
    return item


# DELETE — DELETE /items/{item_id}
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int, db: Session = Depends(get_db)):
    """Menghapus item dari database."""
    item = db.query(Item).filter(Item.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=404, detail="Item tidak ditemukan")

    db.delete(item)
    db.commit()
    return None

"""
Modul Autentikasi JWT
BeebaneLabs - https://beebanelabs.pages.dev
"""

from datetime import datetime, timedelta
from typing import Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy.orm import Session
from database import get_db, User

# ===== KONFIGURASI =====
SECRET_KEY = "ganti-dengan-kunci-rahasia-yang-panjang-dan-acak"  # Gunakan env var!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Password hashing dengan bcrypt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# Skema OAuth2 — menunjuk ke endpoint login
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")


def hash_password(password: str) -> str:
    """Meng-hash password menggunakan bcrypt."""
    return pwd_context.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Memverifikasi password terhadap hash."""
    return pwd_context.verify(plain_password, hashed_password)


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """Membuat JWT access token."""
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
):
    """Dependency: Mendapatkan user yang sedang login dari JWT token."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Token tidak valid",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        # Decode JWT token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    # Cari user di database
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise credentials_exception

    return user

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from database import get_db, User
from auth import (
    hash_password, verify_password,
    create_access_token, get_current_user,
    ACCESS_TOKEN_EXPIRE_MINUTES
)
from pydantic import BaseModel, EmailStr
from datetime import timedelta

router = APIRouter(prefix="/auth", tags=["Autentikasi"])


class UserCreate(BaseModel):
    """Schema untuk registrasi user baru."""
    username: str
    email: str
    password: str


class Token(BaseModel):
    """Schema response JWT token."""
    access_token: str
    token_type: str


@router.post("/register", status_code=status.HTTP_201_CREATED)
async def register(user_data: UserCreate, db: Session = Depends(get_db)):
    """Registrasi user baru."""
    # Cek apakah username sudah ada
    existing = db.query(User).filter(
        User.username == user_data.username
    ).first()
    if existing:
        raise HTTPException(status_code=400, detail="Username sudah digunakan")

    # Buat user baru
    new_user = User(
        username=user_data.username,
        email=user_data.email,
        hashed_password=hash_password(user_data.password)
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)

    return {"message": "Registrasi berhasil", "user_id": new_user.id}


@router.post("/login", response_model=Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    """Login dan mendapatkan JWT token."""
    # Cari user di database
    user = db.query(User).filter(
        User.username == form_data.username
    ).first()

    # Verifikasi user dan password
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Username atau password salah",
            headers={"WWW-Authenticate": "Bearer"},
        )

    # Buat JWT token
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )

    return {"access_token": access_token, "token_type": "bearer"}


@router.get("/me")
async def get_my_profile(current_user: User = Depends(get_current_user)):
    """Mendapatkan profil user yang sedang login (protected endpoint)."""
    return {
        "id": current_user.id,
        "username": current_user.username,
        "email": current_user.email,
        "is_active": current_user.is_active
    }

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import time

app = FastAPI()

# ===== CORS MIDDLEWARE =====
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",     # React dev server
        "http://localhost:5173",     # Vite dev server
        "https://beebanelabs.pages.dev",  # Production frontend
    ],
    allow_credentials=True,   # Izinkan cookies
    allow_methods=["*"],      # Izinkan semua HTTP methods
    allow_headers=["*"],      # Izinkan semua headers
)

# ===== MIDDLEWARE: REQUEST LOGGING & TIMING =====
@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Middleware untuk logging dan mengukur waktu response."""
    start_time = time.time()

    # Proses request
    response = await call_next(request)

    # Hitung durasi
    duration = round((time.time() - start_time) * 1000, 2)

    # Log informasi request
    print(
        f"[{request.method}] {request.url.path} "
        f"→ {response.status_code} ({duration}ms)"
    )

    # Tambahkan header custom
    response.headers["X-Process-Time"] = f"{duration}ms"
    response.headers["X-API-Version"] = "1.0.0"

    return response


# ===== MIDDLEWARE: GLOBAL ERROR HANDLING =====
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Menangkap semua error yang tidak tertangani."""
    print(f"[ERROR] {request.method} {request.url.path}: {str(exc)}")
    return JSONResponse(
        status_code=500,
        content={
            "detail": "Terjadi kesalahan internal server",
            "error": str(exc)  # Hapus di production!
        }
    )

# requirements-prod.txt
fastapi==0.115.0
uvicorn[standard]==0.30.0
gunicorn==22.0.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.9
aiosqlite==0.20.0

# Jalankan dengan Gunicorn (4 worker processes)
gunicorn main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --timeout 120 \
    --access-logfile - \
    --error-logfile -

# Alternatif: langsung dengan Uvicorn (satu proses)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

[Unit]
Description=FastAPI Application
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/opt/fastapi-project
Environment="PATH=/opt/fastapi-project/venv/bin"
Environment="SECRET_KEY=kunci-rahasia-produksi-anda"
Environment="DATABASE_URL=sqlite:///./prod.db"
ExecStart=/opt/fastapi-project/venv/bin/gunicorn main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 127.0.0.1:8000
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

server {
    listen 80;
    server_name api.beebanelabs.pages.dev;

    # Redirect ke HTTPS
    return 301 https://$host$request_uri;
}

server {
    listen 443 ssl http2;
    server_name api.beebanelabs.pages.dev;

    ssl_certificate     /etc/letsencrypt/live/api.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/api.example.com/privkey.pem;

    # Proxy ke FastAPI (Gunicorn + Uvicorn)
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

    # Rate limiting
    limit_req_zone $binary_remote_addr zone=api:10m rate=30r/s;
    location /api/ {
        limit_req zone=api burst=50 nodelay;
        proxy_pass http://127.0.0.1:8000;
    }
}

FROM python:3.11-slim

WORKDIR /app

# Copy dan install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy source code
COPY . .

# Expose port
EXPOSE 8000

# Jalankan dengan Uvicorn
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# Build image
docker build -t fastapi-app .

# Jalankan container
docker run -d -p 8000:8000 \
    -e SECRET_KEY="kunci-rahasia-anda" \
    --name fastapi-app \
    fastapi-app

# Cek status
docker ps
docker logs fastapi-app