API Design Patterns Backend

Thiết kế APIs chuyên nghiệp - consistent, scalable, developer-friendly

Learning Outcomes

Sau khi hoàn thành trang này, bạn sẽ:

• 🎯 Apply REST best practices cho resource-oriented APIs
• 🎯 Implement API versioning strategies phù hợp
• 🎯 Master pagination patterns cho large datasets
• 🎯 Design error handling conventions nhất quán
• 🎯 Implement rate limiting để protect APIs
• 🎯 Tránh các Production Pitfalls phổ biến

---

REST Best Practices

Resource Naming

# ✅ GOOD: Nouns, plural, lowercase, kebab-case
GET    /users              # List users
GET    /users/123          # Get user 123
POST   /users              # Create user
PUT    /users/123          # Update user 123
DELETE /users/123          # Delete user 123

GET    /users/123/posts    # User's posts (nested resource)
GET    /blog-posts         # Kebab-case for multi-word

# ❌ BAD: Verbs, singular, camelCase
GET    /getUser/123        # Verb in URL
POST   /createUser         # Verb in URL
GET    /user/123           # Singular
GET    /blogPosts          # camelCase

HTTP Methods

Method	Purpose	Idempotent	Safe
GET	Read resource	✅	✅
POST	Create resource	❌	❌
PUT	Replace resource	✅	❌
PATCH	Partial update	❌	❌
DELETE	Remove resource	✅	❌

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: str

class UserUpdate(BaseModel):
    name: str
    email: str

class UserPatch(BaseModel):
    name: str | None = None
    email: str | None = None

# GET - Read (Safe, Idempotent)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id, "name": "HPN"}

# POST - Create (Not Idempotent)
@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    return {"id": 1, **user.dict()}

# PUT - Full Replace (Idempotent)
@app.put("/users/{user_id}")
async def replace_user(user_id: int, user: UserUpdate):
    return {"id": user_id, **user.dict()}

# PATCH - Partial Update
@app.patch("/users/{user_id}")
async def update_user(user_id: int, user: UserPatch):
    update_data = user.dict(exclude_unset=True)
    return {"id": user_id, **update_data}

# DELETE - Remove (Idempotent)
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
    return None

HTTP Status Codes

from fastapi import status

# === SUCCESS ===
# 200 OK - General success
# 201 Created - Resource created
# 204 No Content - Success with no body (DELETE)

# === CLIENT ERRORS ===
# 400 Bad Request - Invalid input
# 401 Unauthorized - Authentication required
# 403 Forbidden - Authenticated but not authorized
# 404 Not Found - Resource doesn't exist
# 409 Conflict - Resource conflict (duplicate)
# 422 Unprocessable Entity - Validation error

# === SERVER ERRORS ===
# 500 Internal Server Error - Unexpected error
# 502 Bad Gateway - Upstream service error
# 503 Service Unavailable - Temporarily unavailable

@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    if user_exists(user.email):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Email already registered"
        )
    return new_user

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User {user_id} not found"
        )
    return user

Query Parameters vs Path Parameters

# Path parameters: Identify specific resource
GET /users/123              # User ID 123
GET /users/123/posts/456    # Post 456 of user 123

# Query parameters: Filter, sort, paginate
GET /users?role=admin       # Filter by role
GET /users?sort=name        # Sort by name
GET /users?page=2&limit=10  # Pagination

# ✅ GOOD patterns
GET /users?status=active&role=admin
GET /posts?author_id=123&published=true
GET /products?min_price=100&max_price=500

# ❌ BAD patterns
GET /users/active           # Use query param instead
GET /getUsersByRole/admin   # Verb in URL

---

API Versioning

URL Path Versioning (Recommended)

from fastapi import FastAPI, APIRouter

app = FastAPI()

# Version 1 router
v1_router = APIRouter(prefix="/api/v1")

@v1_router.get("/users")
async def list_users_v1():
    return {"version": "v1", "users": []}

# Version 2 router
v2_router = APIRouter(prefix="/api/v2")

@v2_router.get("/users")
async def list_users_v2():
    # New response format
    return {
        "version": "v2",
        "data": {"users": []},
        "meta": {"total": 0}
    }

app.include_router(v1_router)
app.include_router(v2_router)

# URLs:
# GET /api/v1/users
# GET /api/v2/users

Header Versioning

from fastapi import FastAPI, Header, HTTPException

@app.get("/users")
async def list_users(
    api_version: str = Header(default="v1", alias="X-API-Version")
):
    if api_version == "v1":
        return {"users": []}
    elif api_version == "v2":
        return {"data": {"users": []}, "meta": {}}
    else:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported API version: {api_version}"
        )

# Request:
# GET /users
# X-API-Version: v2

Query Parameter Versioning

@app.get("/users")
async def list_users(version: str = "v1"):
    if version == "v1":
        return {"users": []}
    elif version == "v2":
        return {"data": {"users": []}}

# URL: GET /users?version=v2

Versioning Best Practices

# ✅ GOOD: Semantic versioning for major changes
/api/v1/users  # Original
/api/v2/users  # Breaking changes

# ✅ GOOD: Deprecation headers
@app.get("/api/v1/users")
async def list_users_v1():
    response = JSONResponse(content={"users": []})
    response.headers["Deprecation"] = "true"
    response.headers["Sunset"] = "2025-01-01"
    response.headers["Link"] = '</api/v2/users>; rel="successor-version"'
    return response

# ✅ GOOD: Support multiple versions simultaneously
# Maintain v1 for 6-12 months after v2 release

# ❌ BAD: Too many versions
/api/v1.0.1/users  # Too granular
/api/v15/users     # Too many versions

---

Pagination Patterns

Offset-based Pagination

from fastapi import FastAPI, Query
from pydantic import BaseModel
from typing import List, Generic, TypeVar

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    page: int
    page_size: int
    total_pages: int

@app.get("/users", response_model=PaginatedResponse[User])
async def list_users(
    page: int = Query(1, ge=1, description="Page number"),
    page_size: int = Query(10, ge=1, le=100, description="Items per page")
):
    offset = (page - 1) * page_size
    
    # Query database
    users = db.query(User).offset(offset).limit(page_size).all()
    total = db.query(User).count()
    
    return PaginatedResponse(
        items=users,
        total=total,
        page=page,
        page_size=page_size,
        total_pages=(total + page_size - 1) // page_size
    )

# Response:
# {
#   "items": [...],
#   "total": 100,
#   "page": 2,
#   "page_size": 10,
#   "total_pages": 10
# }

Cursor-based Pagination (Recommended for Large Datasets)

from pydantic import BaseModel
from typing import Optional
import base64

class CursorPaginatedResponse(BaseModel):
    items: list
    next_cursor: Optional[str]
    has_more: bool

def encode_cursor(user_id: int) -> str:
    return base64.b64encode(f"id:{user_id}".encode()).decode()

def decode_cursor(cursor: str) -> int:
    decoded = base64.b64decode(cursor.encode()).decode()
    return int(decoded.split(":")[1])

@app.get("/users")
async def list_users(
    cursor: Optional[str] = None,
    limit: int = Query(10, ge=1, le=100)
):
    query = db.query(User).order_by(User.id)
    
    if cursor:
        last_id = decode_cursor(cursor)
        query = query.filter(User.id > last_id)
    
    users = query.limit(limit + 1).all()  # Fetch one extra
    
    has_more = len(users) > limit
    if has_more:
        users = users[:-1]  # Remove extra item
    
    next_cursor = encode_cursor(users[-1].id) if has_more else None
    
    return CursorPaginatedResponse(
        items=users,
        next_cursor=next_cursor,
        has_more=has_more
    )

# Response:
# {
#   "items": [...],
#   "next_cursor": "aWQ6MTIz",
#   "has_more": true
# }

Pagination Comparison

Feature	Offset	Cursor
Jump to page	✅	❌
Consistent results	❌	✅
Performance (large data)	❌ Slow	✅ Fast
Real-time data	❌ Duplicates/skips	✅ Consistent
Implementation	Simple	Complex

# ✅ Use Offset for:
# - Small datasets (<10K records)
# - Admin dashboards
# - When users need to jump to specific pages

# ✅ Use Cursor for:
# - Large datasets
# - Real-time feeds (social media, notifications)
# - Infinite scroll UIs
# - When data changes frequently

---

Error Handling

Consistent Error Response Format

from pydantic import BaseModel
from typing import Optional, List
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class ErrorDetail(BaseModel):
    field: Optional[str] = None
    message: str
    code: str

class ErrorResponse(BaseModel):
    error: str
    message: str
    details: Optional[List[ErrorDetail]] = None
    request_id: Optional[str] = None

# Custom exception
class APIException(Exception):
    def __init__(
        self,
        status_code: int,
        error: str,
        message: str,
        details: List[ErrorDetail] = None
    ):
        self.status_code = status_code
        self.error = error
        self.message = message
        self.details = details

# Exception handler
@app.exception_handler(APIException)
async def api_exception_handler(request: Request, exc: APIException):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            error=exc.error,
            message=exc.message,
            details=exc.details,
            request_id=getattr(request.state, "request_id", None)
        ).dict()
    )

# Usage
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    if not user:
        raise APIException(
            status_code=404,
            error="USER_NOT_FOUND",
            message=f"User with ID {user_id} does not exist"
        )
    return user

Validation Error Handler

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request,
    exc: RequestValidationError
):
    details = []
    for error in exc.errors():
        details.append(ErrorDetail(
            field=".".join(str(loc) for loc in error["loc"][1:]),  # Skip 'body'
            message=error["msg"],
            code=error["type"]
        ))
    
    return JSONResponse(
        status_code=422,
        content=ErrorResponse(
            error="VALIDATION_ERROR",
            message="Request validation failed",
            details=details
        ).dict()
    )

# Response:
# {
#   "error": "VALIDATION_ERROR",
#   "message": "Request validation failed",
#   "details": [
#     {"field": "email", "message": "invalid email format", "code": "value_error.email"}
#   ]
# }

Error Codes Convention

# Use SCREAMING_SNAKE_CASE for error codes
ERROR_CODES = {
    # Authentication
    "AUTH_REQUIRED": "Authentication is required",
    "AUTH_INVALID_TOKEN": "Invalid or expired token",
    "AUTH_INSUFFICIENT_PERMISSIONS": "Insufficient permissions",
    
    # Resources
    "USER_NOT_FOUND": "User not found",
    "USER_ALREADY_EXISTS": "User already exists",
    "POST_NOT_FOUND": "Post not found",
    
    # Validation
    "VALIDATION_ERROR": "Request validation failed",
    "INVALID_INPUT": "Invalid input data",
    
    # Rate limiting
    "RATE_LIMIT_EXCEEDED": "Too many requests",
    
    # Server
    "INTERNAL_ERROR": "Internal server error",
    "SERVICE_UNAVAILABLE": "Service temporarily unavailable",
}

---

Rate Limiting

Token Bucket Algorithm

from fastapi import FastAPI, Request, HTTPException
from collections import defaultdict
import time

class RateLimiter:
    def __init__(self, requests_per_minute: int = 60):
        self.requests_per_minute = requests_per_minute
        self.tokens = defaultdict(lambda: requests_per_minute)
        self.last_update = defaultdict(time.time)
    
    def is_allowed(self, key: str) -> bool:
        now = time.time()
        time_passed = now - self.last_update[key]
        
        # Refill tokens
        self.tokens[key] = min(
            self.requests_per_minute,
            self.tokens[key] + time_passed * (self.requests_per_minute / 60)
        )
        self.last_update[key] = now
        
        if self.tokens[key] >= 1:
            self.tokens[key] -= 1
            return True
        return False

rate_limiter = RateLimiter(requests_per_minute=60)

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # Use IP or user ID as key
    client_ip = request.client.host
    
    if not rate_limiter.is_allowed(client_ip):
        return JSONResponse(
            status_code=429,
            content={
                "error": "RATE_LIMIT_EXCEEDED",
                "message": "Too many requests. Please try again later."
            },
            headers={
                "Retry-After": "60",
                "X-RateLimit-Limit": "60",
                "X-RateLimit-Remaining": "0"
            }
        )
    
    response = await call_next(request)
    return response

Redis-based Rate Limiting (Production)

import redis
from fastapi import FastAPI, Request, HTTPException

redis_client = redis.Redis(host="localhost", port=6379, db=0)

class RedisRateLimiter:
    def __init__(self, requests_per_minute: int = 60):
        self.limit = requests_per_minute
        self.window = 60  # seconds
    
    def is_allowed(self, key: str) -> tuple[bool, int]:
        pipe = redis_client.pipeline()
        now = int(time.time())
        window_key = f"rate_limit:{key}:{now // self.window}"
        
        pipe.incr(window_key)
        pipe.expire(window_key, self.window)
        results = pipe.execute()
        
        current_count = results[0]
        remaining = max(0, self.limit - current_count)
        
        return current_count <= self.limit, remaining

rate_limiter = RedisRateLimiter(requests_per_minute=100)

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host
    allowed, remaining = rate_limiter.is_allowed(client_ip)
    
    if not allowed:
        return JSONResponse(
            status_code=429,
            content={"error": "RATE_LIMIT_EXCEEDED"},
            headers={
                "X-RateLimit-Limit": "100",
                "X-RateLimit-Remaining": str(remaining),
                "Retry-After": "60"
            }
        )
    
    response = await call_next(request)
    response.headers["X-RateLimit-Limit"] = "100"
    response.headers["X-RateLimit-Remaining"] = str(remaining)
    return response
}

Rate Limit by User Tier

from enum import Enum

class UserTier(Enum):
    FREE = "free"
    PRO = "pro"
    ENTERPRISE = "enterprise"

RATE_LIMITS = {
    UserTier.FREE: 60,        # 60 requests/minute
    UserTier.PRO: 300,        # 300 requests/minute
    UserTier.ENTERPRISE: 1000  # 1000 requests/minute
}

async def get_rate_limit(request: Request) -> int:
    user = await get_current_user(request)
    if user:
        return RATE_LIMITS.get(user.tier, 60)
    return 60  # Default for unauthenticated

---

HATEOAS & Hypermedia

from pydantic import BaseModel
from typing import Dict, Optional

class Link(BaseModel):
    href: str
    method: str = "GET"

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    _links: Dict[str, Link]

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    
    return UserResponse(
        id=user.id,
        name=user.name,
        email=user.email,
        _links={
            "self": Link(href=f"/users/{user_id}"),
            "posts": Link(href=f"/users/{user_id}/posts"),
            "update": Link(href=f"/users/{user_id}", method="PUT"),
            "delete": Link(href=f"/users/{user_id}", method="DELETE"),
        }
    )

# Response:
# {
#   "id": 1,
#   "name": "HPN",
#   "email": "hpn@test.com",
#   "_links": {
#     "self": {"href": "/users/1", "method": "GET"},
#     "posts": {"href": "/users/1/posts", "method": "GET"},
#     "update": {"href": "/users/1", "method": "PUT"},
#     "delete": {"href": "/users/1", "method": "DELETE"}
#   }
# }

---

Production Pitfalls

Pitfall 1: Exposing Internal IDs

# ❌ BUG: Sequential IDs expose business info
GET /users/1
GET /users/2
# Attacker knows you have ~2 users

# ✅ FIX: Use UUIDs or hashids
import uuid
from hashids import Hashids

hashids = Hashids(salt="your-secret-salt", min_length=8)

class User(Base):
    id: int  # Internal
    public_id: str = Field(default_factory=lambda: str(uuid.uuid4()))

# Or encode IDs
def encode_id(id: int) -> str:
    return hashids.encode(id)

def decode_id(hash: str) -> int:
    return hashids.decode(hash)[0]

Pitfall 2: Over-fetching / Under-fetching

# ❌ BAD: Always return everything
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return user  # Returns 50 fields when client needs 3

# ✅ FIX: Field selection
@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    fields: str = Query(None, description="Comma-separated fields")
):
    user = get_user_from_db(user_id)
    
    if fields:
        field_list = fields.split(",")
        return {k: v for k, v in user.dict().items() if k in field_list}
    
    return user

# GET /users/1?fields=id,name,email

Pitfall 3: N+1 in List Endpoints

# ❌ BUG: N+1 queries
@app.get("/posts")
async def list_posts():
    posts = db.query(Post).all()
    return [
        {
            "id": p.id,
            "title": p.title,
            "author": p.author.name  # N queries!
        }
        for p in posts
    ]

# ✅ FIX: Eager loading
from sqlalchemy.orm import joinedload

@app.get("/posts")
async def list_posts():
    posts = db.query(Post).options(joinedload(Post.author)).all()
    return posts

Pitfall 4: Missing Idempotency Keys

# ❌ BUG: Duplicate payments on retry
@app.post("/payments")
async def create_payment(payment: PaymentCreate):
    return process_payment(payment)  # Called twice = charged twice!

# ✅ FIX: Idempotency key
@app.post("/payments")
async def create_payment(
    payment: PaymentCreate,
    idempotency_key: str = Header(...)
):
    # Check if already processed
    existing = get_payment_by_idempotency_key(idempotency_key)
    if existing:
        return existing  # Return cached result
    
    result = process_payment(payment)
    save_idempotency_key(idempotency_key, result)
    return result

Pitfall 5: Inconsistent Response Format

# ❌ BAD: Inconsistent responses
GET /users     → [{"id": 1}, {"id": 2}]
GET /users/1   → {"id": 1, "name": "HPN"}
GET /posts     → {"posts": [...], "count": 10}

# ✅ GOOD: Consistent envelope
class APIResponse(BaseModel):
    data: Any
    meta: Optional[dict] = None
    errors: Optional[list] = None

GET /users     → {"data": [...], "meta": {"total": 2}}
GET /users/1   → {"data": {"id": 1, "name": "HPN"}}
GET /posts     → {"data": [...], "meta": {"total": 10}}

---

Quick Reference

# === REST CONVENTIONS ===
GET    /resources          # List
GET    /resources/{id}     # Get one
POST   /resources          # Create
PUT    /resources/{id}     # Replace
PATCH  /resources/{id}     # Partial update
DELETE /resources/{id}     # Delete

# === STATUS CODES ===
200 OK                     # Success
201 Created                # Resource created
204 No Content             # Success, no body
400 Bad Request            # Invalid input
401 Unauthorized           # Auth required
403 Forbidden              # Not authorized
404 Not Found              # Resource not found
409 Conflict               # Duplicate
422 Unprocessable Entity   # Validation error
429 Too Many Requests      # Rate limited
500 Internal Server Error  # Server error

# === PAGINATION ===
# Offset: ?page=2&page_size=10
# Cursor: ?cursor=abc123&limit=10

# === VERSIONING ===
/api/v1/users              # URL path (recommended)
X-API-Version: v1          # Header
?version=v1                # Query param

# === ERROR FORMAT ===
{
  "error": "ERROR_CODE",
  "message": "Human readable message",
  "details": [{"field": "email", "message": "invalid"}]
}

---

Cross-links

• Prerequisites: FastAPI Deep Dive
• Related: SQLAlchemy & Databases - Database patterns
• See Also: Production Deployment - Deploy APIs
• See Also: Input Validation - Pydantic validation patterns

---