api-design-expert

What this skill does

# 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

```python
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"
)

# Models
class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

class UserCreate(BaseModel):
    email: str = Field(..., example="[email protected]")
    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": "[email protected]",
                "name": "John Doe",
                "role": "user",
                "created_at": "2024-01-01T00:00:00Z",
                "updated_at": "2024-01-01T00:00:00Z"
            }
        }

class UserUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    role: Optional[UserRole] = None

# REST Endpoints following best practices
@app.get("/api/v1/users",
         response_model=List[UserResponse],
         summary="List all users",
         tags=["Users"])
async def list_users(
    page: int = Query(1, ge=1, description="Page number"),
    page_size: int = Query(20, ge=1, le=100, description="Items per page"),
    sort: str = Query("created_at", description="Sort field"),
    order: str = Query("desc", regex="^(asc|desc)$")
):
    """
    Retrieve a paginated list of users.

    - **page**: Page number (starts at 1)
    - **page_size**: Number of items per page (1-100)
    - **sort**: Field to sort by
    - **order**: Sort order (asc or desc)
    """
    # Implementation
    return []

@app.get("/api/v1/users/{user_id}",
         response_model=UserResponse,
         summary="Get user by ID",
         tags=["Users"])
async def get_user(
    user_id: str = Path(..., description="User ID")
):
    """Retrieve a specific user by ID."""
    # Implementation
    raise HTTPException(status_code=404, detail="User not found")

@app.post("/api/v1/users",
          response_model=UserResponse,
          status_code=201,
          summary="Create new user",
          tags=["Users"])
async def create_user(user: UserCreate):
    """Create a new user."""
    # Implementation
    return UserResponse(
        id="123e4567-e89b-12d3-a456-426614174000",
        email=user.email,
        name=user.name,
        role=user.role,
        created_at=datetime.now(),
        updated_at=datetime.now()
    )

@app.patch("/api/v1/users/{user_id}",
           response_model=UserResponse,
           summary="Update user",
           tags=["Users"])
async def update_user(
    user_id: str = Path(..., description="User ID"),
    user: UserUpdate = None
):
    """Partially update a user."""
    # Implementation
    pass

@app.delete("/api/v1/users/{user_id}",
            status_code=204,
            summary="Delete user",
            tags=["Users"])
async def delete_user(
    user_id: str = Path(..., description="User ID")
):
    """Delete a user."""
    # Implementation
    pass

# Nested resources
@app.get("/api/v1/users/{user_id}/posts",
         summary="Get user posts",
         tags=["Users", "Posts"])
async def get_user_posts(user_id: str):
    """Retrieve all posts for a specific user."""
    return []
```

## API Versioning

```python
from fastapi import APIRouter, Request

# URL Path Versioning (Recommended)
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": "[email protected]"}]

# Header Versioning
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()
    return get_users_v1()

# Content Negotiation Versioning
@app.get("/api/users")
async def get_users_content_negotiation(
    accept: str = Header("application/vnd.api+json; version=1")
):
    if "version=2" in accept:
        return get_users_v2()
    return get_users_v1()
```

## Error Handling

```python
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()
    )

# Custom business logic errors
class BusinessError(Exception):
    def __init__(self, message: str, code: str):
        self.message = message
        self.code = code

@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=400,
        content=ErrorResponse(
            error=exc.code,
            message=exc.message,
            timestamp=datetime.now(),
            path=request.url.path
        ).dict()
    )
```

## Rate Limiting

```python
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),