restful-apis

# RESTful API Design — Implementation Guide Covers: **Endpoint Design · HTTP Semantics · Error Handling · Versioning · Documentation · Authentication** ----- ## API Design Principles ### Core REST Constraints REST (Representational State Transfer) is an architectural style built on six constraints. Understanding these constraints is essential for building truly RESTful APIs that are scalable, maintainable, and intuitive. The six constraints are: Client-Server Architecture, which separates concerns between client and server, allowing them to evolve independently; Statelessness, where each request contains all information needed to process it, with no server-side session; Cacheability, enabling clients to cache responses for improved performance; Layered System, allowing intermediaries between client and server; Uniform Interface, providing standardized resource-based interactions; and Code on Demand (optional), where servers can extend client functionality. **Key Principles:** - Resources are the core abstraction — nouns, not verbs - Each resource has a unique URI - Use HTTP methods semantically - Representations describe resource state - Stateless communication between client and server ### Resource Naming ```yaml # Good vs Bad Naming Examples # Bad - Using verbs GET /getUsers # Wrong POST /createUser # Wrong POST /user/create # Wrong GET /getUserById/123 # Wrong # Good - Using nouns GET /users # List users POST /users # Create user GET /users/123 # Get user PATCH /users/123 # Update user DELETE /users/123 # Delete user # Nested resources GET /users/123/orders # User's orders POST /users/123/orders # Create order for user GET /users/123/orders/456 # Specific order GET /users/123/orders/456/items # Items in order # Collections GET /products # All products GET /categories/electronics/products # Products in category # Singleton resources GET /settings # Get settings PATCH /settings # Update settings (not /setting) # Actions as resources POST /users/123/activate # Activate user POST /users/123/deactivate # Deactivate user POST /orders/456/cancel # Cancel order # Compound documents (include related resources) GET /users/123?include=orders,profile ``` ----- ## HTTP Methods and Status Codes ### Proper Method Usage | Method | Safe | Idempotent | Use For | |--------|------|------------|---------| | **GET** | Yes | Yes | Retrieve resources | | **POST** | No | No | Create resources, execute actions | | **PUT** | No | Yes | Replace resources entirely | | **PATCH** | No | Yes | Partial resource update | | **DELETE** | No | Yes | Remove resources | | **HEAD** | Yes | Yes | Get headers only | | **OPTIONS** | Yes | Yes | Get supported methods | ```python from fastapi import FastAPI, HTTPException, status from pydantic import BaseModel, EmailStr from typing import List, Optional from datetime import datetime import uuid app = FastAPI() # In-memory storage users_db = {} # Request/Response models class UserCreate(BaseModel): name: str email: EmailStr age: Optional[int] = None class UserUpdate(BaseModel): name: Optional[str] = None email: Optional[EmailStr] = None age: Optional[int] = None class User(BaseModel): id: str name: str email: EmailStr age: Optional[int] = None created_at: datetime # GET - Retrieve resources @app.get("/users", response_model=List[User]) async def list_users( limit: int = 10, offset: int = 0, search: Optional[str] = None ): """List users with pagination and search""" users = list(users_db.values()) if search: users = [u for u in users if search.lower() in u.name.lower()] return users[offset:offset + limit] @app.get("/users/{user_id}", response_model=User) async def get_user(user_id: str): """Get a specific user""" if user_id not in users_db: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) return users_db[user_id] # POST - Create resources @app.post("/users", response_model=User, status_code=status.HTTP_201_CREATED) async def create_user(user: UserCreate): """Create a new user""" # Check for duplicate email existing = [u for u in users_db.values() if u.email == user.email] if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Email already registered" ) # Create user new_user = User( id=str(uuid.uuid4()), name=user.name, email=user.email, age=user.age, created_at=datetime.now() ) users_db[new_user.id] = new_user return new_user # PUT - Replace entire resource @app.put("/users/{user_id}", response_model=User) async def replace_user(user_id: str, user: UserCreate): """Replace a user (full update)""" if user_id not in users_db: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) updated_user = User( id=user_id, name=user.name, email=user.email, age=user.age, created_at=users_db[user_id].created_at ) users_db[user_id] = updated_user return updated_user # PATCH - Partial update @app.patch("/users/{user_id}", response_model=User) async def update_user(user_id: str, user_update: UserUpdate): """Update user fields (partial update)""" if user_id not in users_db: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) existing_user = users_db[user_id] # Update only provided fields update_data = user_update.dict(exclude_unset=True) for field, value in update_data.items(): setattr(existing_user, field, value) users_db[user_id] = existing_user return existing_user # DELETE - Remove resources @app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_user(user_id: str): """Delete a user""" if user_id not in users_db: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) del users_db[user_id] return None # HEAD - Check resource existence @app.head("/users/{user_id}") async def check_user_exists(user_id: str): """Check if user exists""" if user_id not in users_db: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) return {"Content-Type": "application/json"} # OPTIONS - Get supported methods @app.options("/users") async def list_supported_methods(): """List supported HTTP methods""" return { "GET": {"description": "List users"}, "POST": {"description": "Create user"}, "OPTIONS": {"description": "List supported methods"} } ``` ### HTTP Status Codes ```python # Success Codes 200 OK # GET, PATCH successful 201 Created # POST created new resource 202 Accepted # Async processing started 204 No Content # DELETE successful, no body to return # Redirection Codes 301 Moved Permanently # Resource moved permanently 302 Found # Temporary redirect 304 Not Modified # Cached response still valid # Client Error Codes 400 Bad Request # Invalid request format 401 Unauthorized # Authentication required 403 Forbidden # Authenticated but not authorized 404 Not Found # Resource doesn't exist 405 Method Not Allowed # HTTP method not supported 409 Conflict # Business logic conflict (e.g., duplicate) 422 Unprocessable Entity # Valid format but semantic errors 429 Too Many Requests # Rate limit exceeded # Server Error Codes 500 Internal Server Error # Unexpected server error 502 Bad Gateway # Upstream service error 503 Service Unavailable # Temporary overload 504 Gateway Timeout # Upstream timeout ``` ----- ## Query Parameters ### Filtering, Sorting, Pagination ```python from enum import Enum from typing import List, Optional, Set from pydantic import BaseModel class SortOrder(str, Enum): ASC = "asc" DESC = "desc" class UserFilter(BaseModel): name: Optional[str] = None email: Optional[str] = None min_age: Optional[int] = None max_age: Optional[int] = None is_active: Optional[bool] = None @app.get("/users") async def list_users_filtered( # Pagination page: int = 1, limit: int = 20, # Filtering search: Optional[str] = None, name: Optional[str] = None, email: Optional[str] = None, min_age: Optional[int] = None, max_age: Optional[int] = None, is_active: Optional[bool] = None, role: Optional[str] = None, # Sorting sort_by: str = "created_at", sort_order: SortOrder = SortOrder.DESC, # Field selection fields: Optional[str] = None, # Related resources include: Optional[str] = None ): """List users with full query parameter support""" # Build query query = list(users_db.values()) # Apply filters if search: query = [u for u in query if search.lower() in u.name.lower() or search.lower() in u.email.lower()] if name: query = [u for u in query if name.lower() in u.name.lower()] if email: query = [u for u in query if email.lower() in u.email.lower()] if min_age is not None: query = [u for u in query if u.age and u.age >= min_age] if max_age is not None: query = [u for u in query if u.age and u.age <= max_age] if is_active is not None: query = [u for u in query if u.is_active == is_active] if role: query = [u for u in query if u.role == role] # Apply sorting reverse = sort_order == SortOrder.DESC query = sorted(query, key=lambda x: getattr(x, sort_by, ""), reverse=reverse) # Apply pagination total = len(query) start = (page - 1) * limit end = start + limit items = query[start:end] # Apply field selection if fields: field_set = set(fields.split(",")) items = [ {k: v for k, v in item.dict().items() if k in field_set} for item in items ] # Build response return { "data": items, "pagination": { "page": page, "limit": limit, "total": total, "total_pages": (total + limit - 1) // limit }, "sort": { "by": sort_by, "order": sort_order } } ``` ### Cursor-Based Pagination ```python class CursorPaginator: """Cursor-based pagination for large datasets""" def __init__(self, page_size: int = 20): self.page_size = page_size def paginate( self, query, cursor: Optional[str] = None, sort_by: str = "created_at", sort_order: SortOrder = SortOrder.DESC ): # Decode cursor if cursor: cursor_data = self._decode_cursor(cursor) last_value = cursor_data.get(sort_by) else: last_value = None # Apply cursor filter if last_value is not None: if sort_order == SortOrder.DESC: query = [q for q in query if getattr(q, sort_by) < last_value] else: query = [q for q in query if getattr(q, sort_by) > last_value] # Sort and limit reverse = sort_order == SortOrder.DESC query = sorted(query, key=lambda x: getattr(x, sort_by, ""), reverse=reverse) items = query[:self.page_size + 1] # Get one extra to check if more exist has_more = len(items) > self.page_size items = items[:self.page_size] # Create next cursor next_cursor = None if has_more and items: last_item = items[-1] next_cursor = self._encode_cursor({ sort_by: getattr(last_item, sort_by) }) return { "data": items, "pagination": { "has_more": has_more, "next_cursor": next_cursor } } def _encode_cursor(self, data: dict) -> str: import base64 return base64.b64encode(json.dumps(data).encode()).decode() def _decode_cursor(self, cursor: str) -> dict: import base64 return json.loads(base64.b64decode(cursor.encode()).decode()) ``` ----- ## Error Handling ### Standardized Error Responses ```python from typing import Optional, List, Any, Dict from pydantic import BaseModel from fastapi import Request from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError from starlette.exceptions import HTTPException as StarletteHTTPException # Error response model class ErrorDetail(BaseModel): field: str message: str code: str class ErrorResponse(BaseModel): error: str message: str code: str details: Optional[List[ErrorDetail]] = None request_id: Optional[str] = None timestamp: str help_url: Optional[str] = None @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request: Request, exc: StarletteHTTPException): return JSONResponse( status_code=exc.status_code, content=ErrorResponse( error=exc.__class__.__name__, message=exc.detail, code=f"ERR_{exc.status_code}",