166 lines
4.5 KiB
Python
166 lines
4.5 KiB
Python
"""
|
|
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)
|