""" Production-ready REST API template using FastAPI. Includes pagination, filtering, error handling, and best practices. """ from fastapi import FastAPI, HTTPException, Query, Path, Depends, status from fastapi.responses import JSONResponse from pydantic import BaseModel, Field, EmailStr from typing import Optional, List, Any from datetime import datetime from enum import Enum app = FastAPI( title="API Template", version="1.0.0", docs_url="/api/docs" ) # Models class UserStatus(str, Enum): ACTIVE = "active" INACTIVE = "inactive" SUSPENDED = "suspended" class UserBase(BaseModel): email: EmailStr name: str = Field(..., min_length=1, max_length=100) status: UserStatus = UserStatus.ACTIVE class UserCreate(UserBase): password: str = Field(..., min_length=8) class UserUpdate(BaseModel): email: Optional[EmailStr] = None name: Optional[str] = Field(None, min_length=1, max_length=100) status: Optional[UserStatus] = None class User(UserBase): id: str created_at: datetime updated_at: datetime class Config: from_attributes = True # Pagination class PaginationParams(BaseModel): page: int = Field(1, ge=1) page_size: int = Field(20, ge=1, le=100) class PaginatedResponse(BaseModel): items: List[Any] total: int page: int page_size: int pages: int # Error handling class ErrorDetail(BaseModel): field: Optional[str] = None message: str code: str class ErrorResponse(BaseModel): error: str message: str details: Optional[List[ErrorDetail]] = None @app.exception_handler(HTTPException) async def http_exception_handler(request, exc): return JSONResponse( status_code=exc.status_code, content=ErrorResponse( error=exc.__class__.__name__, message=exc.detail if isinstance(exc.detail, str) else exc.detail.get("message", "Error"), details=exc.detail.get("details") if isinstance(exc.detail, dict) else None ).dict() ) # Endpoints @app.get("/api/users", response_model=PaginatedResponse, tags=["Users"]) async def list_users( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), status: Optional[UserStatus] = Query(None), search: Optional[str] = Query(None) ): """List users with pagination and filtering.""" # Mock implementation total = 100 items = [ User( id=str(i), email=f"user{i}@example.com", name=f"User {i}", status=UserStatus.ACTIVE, created_at=datetime.now(), updated_at=datetime.now() ).dict() for i in range((page-1)*page_size, min(page*page_size, total)) ] return PaginatedResponse( items=items, total=total, page=page, page_size=page_size, pages=(total + page_size - 1) // page_size ) @app.post("/api/users", response_model=User, status_code=status.HTTP_201_CREATED, tags=["Users"]) async def create_user(user: UserCreate): """Create a new user.""" # Mock implementation return User( id="123", email=user.email, name=user.name, status=user.status, created_at=datetime.now(), updated_at=datetime.now() ) @app.get("/api/users/{user_id}", response_model=User, tags=["Users"]) async def get_user(user_id: str = Path(..., description="User ID")): """Get user by ID.""" # Mock: Check if exists if user_id == "999": raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={"message": "User not found", "details": {"id": user_id}} ) return User( id=user_id, email="user@example.com", name="User Name", status=UserStatus.ACTIVE, created_at=datetime.now(), updated_at=datetime.now() ) @app.patch("/api/users/{user_id}", response_model=User, tags=["Users"]) async def update_user(user_id: str, update: UserUpdate): """Partially update user.""" # Validate user exists existing = await get_user(user_id) # Apply updates update_data = update.dict(exclude_unset=True) for field, value in update_data.items(): setattr(existing, field, value) existing.updated_at = datetime.now() return existing @app.delete("/api/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Users"]) async def delete_user(user_id: str): """Delete user.""" await get_user(user_id) # Verify exists return None if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)