| name | api-design-expert |
| version | 1.0.0 |
| description | Expert-level API design principles, REST, GraphQL, versioning, and API best practices |
| category | api |
| tags | ["api-design","rest","graphql","api-versioning","api-security"] |
| allowed-tools | ["Read","Write","Edit"] |
API Design Expert
Expert guidance for API design, RESTful principles, GraphQL, versioning strategies, and API best practices.
Core Concepts
API Design Principles
- RESTful architecture
- Resource-oriented design
- Uniform interface
- Statelessness
- Cacheability
- Layered system
API Styles
- REST (Representational State Transfer)
- GraphQL
- RPC (Remote Procedure Call)
- WebSocket
- Server-Sent Events (SSE)
- gRPC
Key Considerations
- Versioning strategies
- Authentication and authorization
- Rate limiting
- Error handling
- Documentation
- Backward compatibility
REST API Design
from fastapi import FastAPI, HTTPException, Query, Path, Header
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
from enum import Enum
app = FastAPI(
title="User Management API",
version="1.0.0",
description="RESTful API for user management"
)
class UserRole(str, Enum):
ADMIN = "admin"
USER = "user"
GUEST = "guest"
class UserCreate(BaseModel):
email: str = Field(..., example="user@example.com")
name: str = Field(..., min_length=1, max_length=100)
role: UserRole = UserRole.USER
class UserResponse(BaseModel):
id: str
email: str
name: str
role: UserRole
created_at: datetime
updated_at: datetime
class Config:
schema_extra = {
"example": {
"id": "123e4567-e89b-12d3-a456-426614174000",
"email": "user@example.com",
"name": "John Doe",
: ,
: ,
:
}
}
():
name: [] = Field(, min_length=, max_length=)
role: [UserRole] =
():
[]
():
HTTPException(status_code=, detail=)
():
UserResponse(
=,
email=user.email,
name=user.name,
role=user.role,
created_at=datetime.now(),
updated_at=datetime.now()
)
():
():
():
[]
API Versioning
from fastapi import APIRouter, Request
v1_router = APIRouter(prefix="/api/v1")
v2_router = APIRouter(prefix="/api/v2")
@v1_router.get("/users")
async def get_users_v1():
"""Version 1: Returns basic user info"""
return [{"id": 1, "name": "John"}]
@v2_router.get("/users")
async def get_users_v2():
"""Version 2: Returns enhanced user info"""
return [{"id": 1, "name": "John", "email": "john@example.com"}]
async def version_from_header(request: Request):
version = request.headers.get("API-Version", "1")
return version
@app.get("/api/users")
async def get_users(version: str = Depends(version_from_header)):
if version == "2":
return get_users_v2()
get_users_v1()
():
accept:
get_users_v2()
get_users_v1()
Error Handling
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ValidationError
class ErrorResponse(BaseModel):
error: str
message: str
details: Optional[dict] = None
timestamp: datetime
path: str
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content=ErrorResponse(
error=exc.status_code,
message=exc.detail,
timestamp=datetime.now(),
path=request.url.path
).dict()
)
@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
return JSONResponse(
status_code=422,
content=ErrorResponse(
error="validation_error",
message="Request validation failed",
details=exc.errors(),
timestamp=datetime.now(),
path=request.url.path
).dict()
)
class BusinessError(Exception):
def __init__(self, message: , code: ):
.message = message
.code = code
():
JSONResponse(
status_code=,
content=ErrorResponse(
error=exc.code,
message=exc.message,
timestamp=datetime.now(),
path=request.url.path
).()
)
Rate Limiting
from fastapi import Request, HTTPException
from datetime import datetime, timedelta
import redis
from typing import Dict
class RateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
async def check_rate_limit(self,
key: str,
max_requests: int,
window_seconds: int) -> Dict:
"""
Token bucket algorithm for rate limiting
"""
now = datetime.now().timestamp()
window_key = f"rate_limit:{key}:{int(now // window_seconds)}"
pipe = self.redis.pipeline()
pipe.incr(window_key)
pipe.expire(window_key, window_seconds)
result = pipe.execute()
request_count = result[0]
if request_count > max_requests:
reset_time = (int(now // window_seconds) + 1) * window_seconds
raise HTTPException(
status_code=429,
detail="Rate limit exceeded",
headers={
"X-RateLimit-Limit": str(max_requests),
"X-RateLimit-Remaining": "0",
"X-RateLimit-Reset": str(reset_time)
}
)
{
: max_requests,
: max_requests - request_count,
: ((now // window_seconds) + ) * window_seconds
}
():
limiter = RateLimiter(redis_client)
client_id = request.client.host
:
rate_info = limiter.check_rate_limit(
client_id,
max_requests=,
window_seconds=
)
response = call_next(request)
response.headers[] = (rate_info[])
response.headers[] = (rate_info[])
response.headers[] = (rate_info[])
response
HTTPException e:
JSONResponse(
status_code=e.status_code,
content={: e.detail},
headers=e.headers
)
HATEOAS Implementation
from typing import Dict, List
class HATEOASResponse(BaseModel):
data: dict
links: Dict[str, str]
def create_links(resource_id: str, resource_type: str) -> Dict[str, str]:
"""Create HATEOAS links for a resource"""
return {
"self": f"/api/v1/{resource_type}/{resource_id}",
"update": f"/api/v1/{resource_type}/{resource_id}",
"delete": f"/api/v1/{resource_type}/{resource_id}",
"collection": f"/api/v1/{resource_type}"
}
@app.get("/api/v1/users/{user_id}", response_model=HATEOASResponse)
async def get_user_with_links(user_id: str):
user = get_user_from_db(user_id)
return HATEOASResponse(
data=user,
links={
"self": f"/api/v1/users/{user_id}",
"posts": f"/api/v1/users/{user_id}/posts",
: ,
: ,
:
}
)
API Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from passlib.context import CryptContext
security = HTTPBearer()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Verify JWT token"""
try:
payload = jwt.decode(
credentials.credentials,
SECRET_KEY,
algorithms=[ALGORITHM]
)
return payload
except JWTError:
raise HTTPException(
status_code=401,
detail="Invalid authentication credentials"
)
@app.get("/api/v1/protected")
async def protected_route(token_payload: dict = Depends(verify_token)):
return {"message": "Access granted", "user": token_payload}
async def verify_api_key(api_key: str = Header(...)):
"""Verify API key"""
if api_key not valid_api_keys:
HTTPException(status_code=, detail=)
api_key
Best Practices
Design
- Use nouns for resources, not verbs
- Use plural names for collections
- Use HTTP methods correctly (GET, POST, PUT, PATCH, DELETE)
- Return appropriate status codes
- Support filtering, sorting, and pagination
- Use consistent naming conventions
- Version APIs from the start
Documentation
- Use OpenAPI/Swagger
- Provide example requests and responses
- Document error codes and messages
- Include authentication requirements
- Keep documentation up-to-date
- Provide SDKs when possible
Performance
- Implement caching (ETags, Cache-Control)
- Support compression (gzip)
- Paginate large result sets
- Use async operations for long tasks
- Implement rate limiting
- Monitor API performance
Security
- Use HTTPS always
- Implement authentication and authorization
- Validate all inputs
- Rate limit to prevent abuse
- Log security events
- Use API keys or OAuth 2.0
- Implement CORS properly
Anti-Patterns
❌ Using GET for state-changing operations
❌ Returning inconsistent response formats
❌ No versioning strategy
❌ Poor error messages
❌ No rate limiting
❌ Exposing internal implementation details
❌ Breaking changes without versioning
Resources