Master REST and GraphQL API design principles to build intuitive, scalable, and maintainable APIs that delight developers. Use when designing new APIs, reviewing API specifications, or establishing API design standards.
Master REST and GraphQL API design principles to build intuitive, scalable, and maintainable APIs that delight developers and stand the test of time.
Resource-Oriented Architecture
HTTP Methods Semantics:
GET: Retrieve resources (idempotent, safe)POST: Create new resourcesPUT: Replace entire resource (idempotent)PATCH: Partial resource updatesDELETE: Remove resources (idempotent)Schema-First Development
Query Structure:
URL Versioning:
/api/v1/users
/api/v2/users
Header Versioning:
Accept: application/vnd.api+json; version=1
Query Parameter Versioning:
/api/users?version=1
# Good: Resource-oriented endpoints
GET /api/users # List users (with pagination)
POST /api/users # Create user
GET /api/users/{id} # Get specific user
PUT /api/users/{id} # Replace user
PATCH /api/users/{id} # Update user fields
DELETE /api/users/{id} # Delete user
# Nested resources
GET /api/users/{id}/orders # Get user's orders
POST /api/users/{id}/orders # Create order for user
# Bad: Action-oriented endpoints (avoid)
POST /api/createUser
POST /api/getUserById
POST /api/deleteUser
from typing import List, Optional
from pydantic import BaseModel, Field
class PaginationParams(BaseModel):
page: int = Field(1, ge=1, description="Page number")
page_size: int = Field(20, ge=1, le=100, description="Items per page")
class FilterParams(BaseModel):
status: Optional[str] = None
created_after: Optional[str] = None
search: Optional[str] = None
class PaginatedResponse(BaseModel):
items: List[dict]
total: int
page: int
page_size: int
pages: int
@property
def has_next(self) -> bool:
return self.page < self.pages
@property
def has_prev(self) -> bool:
return self.page > 1
# FastAPI endpoint example
from fastapi import FastAPI, Query, Depends
app = FastAPI()
@app.get("/api/users", response_model=PaginatedResponse)
async def list_users(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
status: Optional[str] = Query(None),
search: Optional[str] = Query(None)
):
# Apply filters
query = build_query(status=status, search=search)
# Count total
total = await count_users(query)
# Fetch page
offset = (page - 1) * page_size
users = await fetch_users(query, limit=page_size, offset=offset)
return PaginatedResponse(
items=users,
total=total,
page=page,
page_size=page_size,
pages=(total + page_size - 1) // page_size
)
from fastapi import HTTPException, status
from pydantic import BaseModel
class ErrorResponse(BaseModel):
error: str
message: str
details: Optional[dict] = None
timestamp: str
path: str
class ValidationErrorDetail(BaseModel):
field: str
message: str
value: Any
# Consistent error responses
STATUS_CODES = {
"success": 200,
"created": 201,
"no_content": 204,
"bad_request": 400,
"unauthorized": 401,
"forbidden": 403,
"not_found": 404,
"conflict": 409,
"unprocessable": 422,
"internal_error": 500
}
def raise_not_found(resource: str, id: str):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={
"error": "NotFound",
"message": f"{resource} not found",
"details": {"id": id}
}
)
def raise_validation_error(errors: List[ValidationErrorDetail]):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={
"error": "ValidationError",
"message": "Request validation failed",
"details": {"errors": [e.dict() for e in errors]}
}
)
# Example usage
@app.get("/api/users/{user_id}")
async def get_user(user_id: str):
user = await fetch_user(user_id)
if not user:
raise_not_found("User", user_id)
return user
class UserResponse(BaseModel):
id: str
name: str
email: str
_links: dict
@classmethod
def from_user(cls, user: User, base_url: str):
return cls(
id=user.id,
name=user.name,
email=user.email,
_links={
"self": {"href": f"{base_url}/api/users/{user.id}"},
"orders": {"href": f"{base_url}/api/users/{user.id}/orders"},
"update": {
"href": f"{base_url}/api/users/{user.id}",
"method": "PATCH"
},
"delete": {
"href": f"{base_url}/api/users/{user.id}",
"method": "DELETE"
}
}
)
# Good - Plural nouns
GET /api/users
GET /api/orders
GET /api/products
# Bad - Verbs or mixed conventions
GET /api/getUser
GET /api/user (inconsistent singular)
POST /api/createOrder
# Shallow nesting (preferred)
GET /api/users/{id}/orders
GET /api/orders/{id}
# Deep nesting (avoid)
GET /api/users/{id}/orders/{orderId}/items/{itemId}/reviews
# Better:
GET /api/order-items/{id}/reviews
GET /api/users → 200 OK (with list)
GET /api/users/{id} → 200 OK or 404 Not Found
GET /api/users?page=2 → 200 OK (paginated)
POST /api/users
Body: {"name": "John", "email": "[email protected]"}
→ 201 Created
Location: /api/users/123
Body: {"id": "123", "name": "John", ...}
POST /api/users (validation error)
→ 422 Unprocessable Entity
Body: {"errors": [...]}
PUT /api/users/{id}
Body: {complete user object}
→ 200 OK (updated)
→ 404 Not Found (doesn't exist)
# Must include ALL fields
PATCH /api/users/{id}
Body: {"name": "Jane"} (only changed fields)
→ 200 OK
→ 404 Not Found
DELETE /api/users/{id}
→ 204 No Content (deleted)
→ 404 Not Found
→ 409 Conflict (can't delete due to references)
# Filtering
GET /api/users?status=active
GET /api/users?role=admin&status=active
# Sorting
GET /api/users?sort=created_at
GET /api/users?sort=-created_at (descending)
GET /api/users?sort=name,created_at
# Searching
GET /api/users?search=john
GET /api/users?q=john
# Field selection (sparse fieldsets)
GET /api/users?fields=id,name,email
GET /api/users?page=2&page_size=20
Response:
{
"items": [...],
"page": 2,
"page_size": 20,
"total": 150,
"pages": 8
}
GET /api/users?limit=20&cursor=eyJpZCI6MTIzfQ
Response:
{
"items": [...],
"next_cursor": "eyJpZCI6MTQzfQ",
"has_more": true
}
GET /api/users?page=2
Response Headers:
Link: <https://api.example.com/users?page=3>; rel="next",
<https://api.example.com/users?page=1>; rel="prev",
<https://api.example.com/users?page=1>; rel="first",
<https://api.example.com/users?page=8>; rel="last"
/api/v1/users
/api/v2/users
Pros: Clear, easy to route
Cons: Multiple URLs for same resource
GET /api/users
Accept: application/vnd.api+json; version=2
Pros: Clean URLs
Cons: Less visible, harder to test
GET /api/users?version=2
Pros: Easy to test
Cons: Optional parameter can be forgotten
X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 742
X-RateLimit-Reset: 1640000000
Response when limited:
429 Too Many Requests
Retry-After: 3600
from fastapi import HTTPException, Request
from datetime import datetime, timedelta
class RateLimiter:
def __init__(self, calls: int, period: int):
self.calls = calls
self.period = period
self.cache = {}
def check(self, key: str) -> bool:
now = datetime.now()
if key not in self.cache:
self.cache[key] = []
# Remove old requests
self.cache[key] = [
ts for ts in self.cache[key]
if now - ts < timedelta(seconds=self.period)
]
if len(self.cache[key]) >= self.calls:
return False
self.cache[key].append(now)
return True
limiter = RateLimiter(calls=100, period=60)
@app.get("/api/users")
async def get_users(request: Request):
if not limiter.check(request.client.host):
raise HTTPException(
status_code=429,
headers={"Retry-After": "60"}
)
return {"users": [...]}
Authorization: Bearer eyJhbGciOiJIUzI1NiIs...
401 Unauthorized - Missing/invalid token
403 Forbidden - Valid token, insufficient permissions
X-API-Key: your-api-key-here
{
"error": {
"code": "VALIDATION_ERROR",
"message": "Request validation failed",
"details": [
{
"field": "email",
"message": "Invalid email format",
"value": "not-an-email"
}
],
"timestamp": "2025-10-16T12:00:00Z",
"path": "/api/users"
}
}
200 OK: Successful GET, PATCH, PUT201 Created: Successful POST204 No Content: Successful DELETE400 Bad Request: Malformed request401 Unauthorized: Authentication required403 Forbidden: Authenticated but not authorized404 Not Found: Resource doesn't exist409 Conflict: State conflict (duplicate email, etc.)422 Unprocessable Entity: Validation errors429 Too Many Requests: Rate limited500 Internal Server Error: Server error503 Service Unavailable: Temporary downtime# Client caching
Cache-Control: public, max-age=3600
# No caching
Cache-Control: no-cache, no-store, must-revalidate
# Conditional requests
ETag: "33a64df551425fcc55e4d42a148795d9f25f89d4"
If-None-Match: "33a64df551425fcc55e4d42a148795d9f25f89d4"
→ 304 Not Modified
POST /api/users/batch
{
"items": [
{"name": "User1", "email": "[email protected]"},
{"name": "User2", "email": "[email protected]"}
]
}
Response:
{
"results": [
{"id": "1", "status": "created"},
{"id": null, "status": "failed", "error": "Email already exists"}
]
}
POST /api/orders
Idempotency-Key: unique-key-123
If duplicate request:
→ 200 OK (return cached response)
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
from fastapi import FastAPI
app = FastAPI(
title="My API",
description="API for managing users",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
@app.get(
"/api/users/{user_id}",
summary="Get user by ID",
response_description="User details",
tags=["Users"]
)
async def get_user(
user_id: str = Path(..., description="The user ID")
):
"""
Retrieve user by ID.
Returns full user profile including:
- Basic information
- Contact details
- Account status
"""
pass
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"version": "1.0.0",
"timestamp": datetime.now().isoformat()
}
@app.get("/health/detailed")
async def detailed_health():
return {
"status": "healthy",
"checks": {
"database": await check_database(),
"redis": await check_redis(),
"external_api": await check_external_api()
}
}
# schema.graphql
# Clear type definitions
type User {
id: ID!
email: String!
name: String!
createdAt: DateTime!
# Relationships
orders(first: Int = 20, after: String, status: OrderStatus): OrderConnection!
profile: UserProfile
}
type Order {
id: ID!
status: OrderStatus!
total: Money!
items: [OrderItem!]!
createdAt: DateTime!
# Back-reference
user: User!
}
# Pagination pattern (Relay-style)
type OrderConnection {
edges: [OrderEdge!]!
pageInfo: PageInfo!
totalCount: Int!
}
type OrderEdge {
node: Order!
cursor: String!
}
type PageInfo {
hasNextPage: Boolean!
hasPreviousPage: Boolean!
startCursor: String
endCursor: String
}
# Enums for type safety
enum OrderStatus {
PENDING
CONFIRMED
SHIPPED
DELIVERED
CANCELLED
}
# Custom scalars
scalar DateTime
scalar Money
# Query root
type Query {
user(id: ID!): User
users(first: Int = 20, after: String, search: String): UserConnection!
order(id: ID!): Order
}
# Mutation root
type Mutation {
createUser(input: CreateUserInput!): CreateUserPayload!
updateUser(input: UpdateUserInput!): UpdateUserPayload!
deleteUser(id: ID!): DeleteUserPayload!
createOrder(input: CreateOrderInput!): CreateOrderPayload!
}
# Input types for mutations
input CreateUserInput {
email: String!
name: String!
password: String!
}
# Payload types for mutations
type CreateUserPayload {
user: User
errors: [Error!]
}
type Error {
field: String
message: String!
}
from typing import Optional, List
from ariadne import QueryType, MutationType, ObjectType
from dataclasses import dataclass
query = QueryType()
mutation = MutationType()
user_type = ObjectType("User")
@query.field("user")
async def resolve_user(obj, info, id: str) -> Optional[dict]:
"""Resolve single user by ID."""
return await fetch_user_by_id(id)
@query.field("users")
async def resolve_users(
obj,
info,
first: int = 20,
after: Optional[str] = None,
search: Optional[str] = None
) -> dict:
"""Resolve paginated user list."""
# Decode cursor
offset = decode_cursor(after) if after else 0
# Fetch users
users = await fetch_users(
limit=first + 1, # Fetch one extra to check hasNextPage
offset=offset,
search=search
)
# Pagination
has_next = len(users) > first
if has_next:
users = users[:first]
edges = [
{
"node": user,
"cursor": encode_cursor(offset + i)
}
for i, user in enumerate(users)
]
return {
"edges": edges,
"pageInfo": {
"hasNextPage": has_next,
"hasPreviousPage": offset > 0,
"startCursor": edges[0]["cursor"] if edges else None,
"endCursor": edges[-1]["cursor"] if edges else None
},
"totalCount": await count_users(search=search)
}
@user_type.field("orders")
async def resolve_user_orders(user: dict, info, first: int = 20) -> dict:
"""Resolve user's orders (N+1 prevention with DataLoader)."""
# Use DataLoader to batch requests
loader = info.context["loaders"]["orders_by_user"]
orders = await loader.load(user["id"])
return paginate_orders(orders, first)
@mutation.field("createUser")
async def resolve_create_user(obj, info, input: dict) -> dict:
"""Create new user."""
try:
# Validate input
validate_user_input(input)
# Create user
user = await create_user(
email=input["email"],
name=input["name"],
password=hash_password(input["password"])
)
return {
"user": user,
"errors": []
}
except ValidationError as e:
return {
"user": None,
"errors": [{"field": e.field, "message": e.message}]
}
from aiodataloader import DataLoader
from typing import List, Optional
class UserLoader(DataLoader):
"""Batch load users by ID."""
async def batch_load_fn(self, user_ids: List[str]) -> List[Optional[dict]]:
"""Load multiple users in single query."""
users = await fetch_users_by_ids(user_ids)
# Map results back to input order
user_map = {user["id"]: user for user in users}
return [user_map.get(user_id) for user_id in user_ids]
class OrdersByUserLoader(DataLoader):
"""Batch load orders by user ID."""
async def batch_load_fn(self, user_ids: List[str]) -> List[List[dict]]:
"""Load orders for multiple users in single query."""
orders = await fetch_orders_by_user_ids(user_ids)
# Group orders by user_id
orders_by_user = {}
for order in orders:
user_id = order["user_id"]
if user_id not in orders_by_user:
orders_by_user[user_id] = []
orders_by_user[user_id].append(order)
# Return in input order
return [orders_by_user.get(user_id, []) for user_id in user_ids]
# Context setup
def create_context():
return {
"loaders": {
"user": UserLoader(),
"orders_by_user": OrdersByUserLoader()
}
}
# user.graphql
type User {
id: ID!
email: String!
name: String!
posts: [Post!]!
}
extend type Query {
user(id: ID!): User
users(first: Int, after: String): UserConnection!
}
extend type Mutation {
createUser(input: CreateUserInput!): CreateUserPayload!
}
# post.graphql
type Post {
id: ID!
title: String!
content: String!
author: User!
}
extend type Query {
post(id: ID!): Post
}
type User {
id: ID! # Always required
email: String! # Required
phone: String # Optional (nullable)
posts: [Post!]! # Non-null array of non-null posts
tags: [String!] # Nullable array of non-null strings
}
interface Node {
id: ID!
createdAt: DateTime!
}
type User implements Node {
id: ID!
createdAt: DateTime!
email: String!
}
type Post implements Node {
id: ID!
createdAt: DateTime!
title: String!
}
type Query {
node(id: ID!): Node
}
union SearchResult = User | Post | Comment
type Query {
search(query: String!): [SearchResult!]!
}
# Query example
{
search(query: "graphql") {
... on User {
name
email
}
... on Post {
title
content
}
... on Comment {
text
author {
name
}
}
}
}
input CreateUserInput {
email: String!
name: String!
password: String!
profileInput: ProfileInput
}
input ProfileInput {
bio: String
avatar: String
website: String
}
input UpdateUserInput {
id: ID!
email: String
name: String
profileInput: ProfileInput
}
type UserConnection {
edges: [UserEdge!]!
pageInfo: PageInfo!
totalCount: Int!
}
type UserEdge {
node: User!
cursor: String!
}
type PageInfo {
hasNextPage: Boolean!
hasPreviousPage: Boolean!
startCursor: String
endCursor: String
}
type Query {
users(first: Int, after: String, last: Int, before: String): UserConnection!
}
# Usage
{
users(first: 10, after: "cursor123") {
edges {
cursor
node {
id
name
}
}
pageInfo {
hasNextPage
endCursor
}
}
}
type UserList {
items: [User!]!
total: Int!
page: Int!
pageSize: Int!
}
type Query {
users(page: Int = 1, pageSize: Int = 20): UserList!
}
input CreatePostInput {
title: String!
content: String!
tags: [String!]
}
type CreatePostPayload {
post: Post
errors: [Error!]
success: Boolean!
}
type Error {
field: String
message: String!
code: String!
}
type Mutation {
createPost(input: CreatePostInput!): CreatePostPayload!
}
type UpdateUserPayload {
user: User
clientMutationId: String
errors: [Error!]
}
input UpdateUserInput {
id: ID!
name: String
clientMutationId: String
}
type Mutation {
updateUser(input: UpdateUserInput!): UpdateUserPayload!
}
input BatchCreateUserInput {
users: [CreateUserInput!]!
}
type BatchCreateUserPayload {
results: [CreateUserResult!]!
successCount: Int!
errorCount: Int!
}
type CreateUserResult {
user: User
errors: [Error!]
index: Int!
}
type Mutation {
batchCreateUsers(input: BatchCreateUserInput!): BatchCreateUserPayload!
}
type Query {
posts(
# Pagination
first: Int = 20
after: String
# Filtering
status: PostStatus
authorId: ID
tag: String
# Sorting
orderBy: PostOrderBy = CREATED_AT
orderDirection: OrderDirection = DESC
# Searching
search: String
): PostConnection!
}
enum PostStatus {
DRAFT
PUBLISHED
ARCHIVED
}
enum PostOrderBy {
CREATED_AT
UPDATED_AT
TITLE
}
enum OrderDirection {
ASC
DESC
}
type User {
firstName: String!
lastName: String!
fullName: String! # Computed in resolver
posts: [Post!]!
postCount: Int! # Computed, doesn't load all posts
}
type Post {
likeCount: Int!
commentCount: Int!
isLikedByViewer: Boolean! # Context-dependent
}
type Subscription {
postAdded: Post!
postUpdated(postId: ID!): Post!
userStatusChanged(userId: ID!): UserStatus!
}
type UserStatus {
userId: ID!
online: Boolean!
lastSeen: DateTime!
}
# Client usage
subscription {
postAdded {
id
title
author {
name
}
}
}
scalar DateTime
scalar Email
scalar URL
scalar JSON
scalar Money
type User {
email: Email!
website: URL
createdAt: DateTime!
metadata: JSON
}
type Product {
price: Money!
}
type User {
name: String!
email: String! @deprecated(reason: "Use emails field instead")
emails: [String!]!
# Conditional inclusion
privateData: PrivateData @include(if: $isOwner)
}
# Query
query GetUser($isOwner: Boolean!) {
user(id: "123") {
name
privateData @include(if: $isOwner) {
ssn
}
}
}
directive @auth(requires: Role = USER) on FIELD_DEFINITION
enum Role {
USER
ADMIN
MODERATOR
}
type Mutation {
deleteUser(id: ID!): Boolean! @auth(requires: ADMIN)
updateProfile(input: ProfileInput!): User! @auth
}
type User {
id: ID!
email: String!
}
type ValidationError {
field: String!
message: String!
}
type NotFoundError {
message: String!
resourceType: String!
resourceId: ID!
}
type AuthorizationError {
message: String!
}
union UserResult = User | ValidationError | NotFoundError | AuthorizationError
type Query {
user(id: ID!): UserResult!
}
# Usage
{
user(id: "123") {
... on User {
id
email
}
... on NotFoundError {
message
resourceType
}
... on AuthorizationError {
message
}
}
}
type CreateUserPayload {
user: User
errors: [Error!]
success: Boolean!
}
type Error {
field: String
message: String!
code: ErrorCode!
}
enum ErrorCode {
VALIDATION_ERROR
UNAUTHORIZED
NOT_FOUND
INTERNAL_ERROR
}
from aiodataloader import DataLoader
class PostLoader(DataLoader):
async def batch_load_fn(self, post_ids):
posts = await db.posts.find({"id": {"$in": post_ids}})
post_map = {post["id"]: post for post in posts}
return [post_map.get(pid) for pid in post_ids]
# Resolver
@user_type.field("posts")
async def resolve_posts(user, info):
loader = info.context["loaders"]["post"]
return await loader.load_many(user["post_ids"])
from graphql import GraphQLError
def depth_limit_validator(max_depth: int):
def validate(context, node, ancestors):
depth = len(ancestors)
if depth > max_depth:
raise GraphQLError(
f"Query depth {depth} exceeds maximum {max_depth}"
)
return validate
def complexity_limit_validator(max_complexity: int):
def calculate_complexity(node):
# Each field = 1, lists multiply
complexity = 1
if is_list_field(node):
complexity *= get_list_size_arg(node)
return complexity
return validate_complexity
type User {
name: String! @deprecated(reason: "Use firstName and lastName")
firstName: String!
lastName: String!
}
# v1 - Initial
type User {
name: String!
}
# v2 - Add optional field (backward compatible)
type User {
name: String!
email: String
}
# v3 - Deprecate and add new field
type User {
name: String! @deprecated(reason: "Use firstName/lastName")
firstName: String!
lastName: String!
email: String
}
@deprecated instead of removing fields/users, not /user)@deprecated directive for gradual migration"""
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.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, EmailStr, ConfigDict
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"
)
# Security Middleware
# Trusted Host: Prevents HTTP Host Header attacks
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["*"] # TODO: Configure this in production, e.g. ["api.example.com"]
)
# CORS: Configures Cross-Origin Resource Sharing
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # TODO: Update this with specific origins in production
allow_credentials=False, # TODO: Set to True if you need cookies/auth headers, but restrict origins
allow_methods=["*"],
allow_headers=["*"],
)
# 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
model_config = ConfigDict(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
).model_dump()
)
# 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()
).model_dump()
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="[email protected]",
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.model_dump(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)