Tutorial FastAPI #3: Cara Mengelola Request Body dan Response Model dengan Pydantic

# Install dependencies
pip install fastapi[all] pydantic email-validator

# Struktur project
myapp/
├── app/
│   ├── models/
│   ├── schemas/
│   └── main.py
└── requirements.txt

from pydantic import BaseModel, EmailStr, constr
from typing import Optional, List
from datetime import datetime

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str
    age: Optional[int] = None

class UserCreate(UserBase):
    # Password minimal 8 karakter
    password: constr(min_length=8) 
    confirm_password: str

    # Custom validator
    @validator('confirm_password')
    def passwords_match(cls, v, values, **kwargs):
        if 'password' in values and v != values['password']:
            raise ValueError('Password tidak cocok!')
        return v

from pydantic import Field, validator

class Product(BaseModel):
    name: str = Field(..., min_length=3, max_length=50)
    price: float = Field(..., gt=0)
    description: Optional[str] = Field(None, max_length=1000)
    tags: List[str] = Field(default_factory=list)

    @validator('name')
    def name_must_be_valid(cls, v):
        if v.strip() == '':
            raise ValueError('Nama produk tidak boleh kosong!')
        return v.strip()

    @validator('price')
    def price_must_be_reasonable(cls, v):
        if v > 1_000_000_000:  # 1 Miliar
            raise ValueError('Harga terlalu tinggi!')
        return v

class BaseResponse(BaseModel):
    success: bool
    message: str
    data: Optional[dict] = None

@app.get(
    "/products/{product_id}",
    response_model=BaseResponse,
    description="Mengambil detail produk berdasarkan ID"
)
async def get_product(product_id: int):
    product = {
        "id": product_id,
        "name": "Laptop Gaming",
        "price": 15_000_000
    }

    return BaseResponse(
        success=True,
        message="Produk berhasil diambil",
        data=product
    )

class ProductList(BaseModel):
    total: int
    products: List[Product]
    page: int
    limit: int
    next_page: Optional[int]
    prev_page: Optional[int]

@app.get(
    "/products/",
    response_model=ProductList,
    description="Mengambil daftar produk dengan pagination"
)
async def list_products(
    page: int = Query(1, ge=1),
    limit: int = Query(10, ge=1, le=100)
):
    # Simulasi data
    products = [
        Product(
            name="Laptop Gaming",
            price=15_000_000,
            tags=["electronics", "gaming"]
        ),
        Product(
            name="Mechanical Keyboard",
            price=1_500_000,
            tags=["accessories"]
        )
    ]

    total = len(products)

    return ProductList(
        total=total,
        products=products,
        page=page,
        limit=limit,
        next_page=page + 1 if total > page * limit else None,
        prev_page=page - 1 if page > 1 else None
    )

class ErrorResponse(BaseModel):
    success: bool = False
    message: str
    error_code: str
    details: Optional[dict] = None

class CustomHTTPException(HTTPException):
    def __init__(
        self,
        status_code: int,
        message: str,
        error_code: str,
        details: Optional[dict] = None
    ):
        super().__init__(
            status_code=status_code,
            detail={
                "success": False,
                "message": message,
                "error_code": error_code,
                "details": details
            }
        )

@app.exception_handler(CustomHTTPException)
async def custom_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content=exc.detail
    )

# Contoh penggunaan
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id <= 0:
        raise CustomHTTPException(
            status_code=400,
            message="User ID tidak valid",
            error_code="INVALID_USER_ID",
            details={"provided_id": user_id}
        )
    # Process valid user_id...

class Address(BaseModel):
    street: str
    city: str
    country: str
    postal_code: str

class User(BaseModel):
    username: str
    addresses: List[Address]

class SearchFilter(BaseModel):
    keyword: str
    category: Optional[str] = None
    min_price: float = Field(0, ge=0)
    max_price: Optional[float] = None
    sort_by: str = Field("date", regex="^(date|price|name)$")
    sort_order: str = Field("desc", regex="^(asc|desc)$")
    page: int = Field(1, ge=1)
    limit: int = Field(10, ge=1, le=100)

class Registration(BaseModel):
    username: str
    password: str
    birth_year: int

    @validator('username')
    def username_alphanumeric(cls, v):
        assert v.isalnum(), 'Username hanya boleh huruf dan angka'
        return v

    @validator('birth_year')
    def valid_birth_year(cls, v):
        current_year = datetime.now().year
        assert 1900 <= v <= current_year, f'Tahun lahir harus antara 1900 dan {current_year}'
        return v