API Design

← Back to System Design 101 | ← Previous: Message Queues

Introduction

Well-designed APIs are the contracts between services and clients. Poor API design leads to frustrated developers, brittle integrations, and maintenance nightmares. Good API design is invisible—it just works.

This article covers API design patterns I've used across REST, GraphQL, and gRPC implementations in production.

REST API Design

REST remains my default choice for most public-facing APIs.

Resource Naming

# Good API design - resource-oriented URLs
GET    /api/v1/users                 # List users
GET    /api/v1/users/{id}            # Get specific user
POST   /api/v1/users                 # Create user
PUT    /api/v1/users/{id}            # Update user (full)
PATCH  /api/v1/users/{id}            # Update user (partial)
DELETE /api/v1/users/{id}            # Delete user

# Nested resources
GET    /api/v1/users/{id}/orders     # List user's orders
GET    /api/v1/users/{id}/orders/{order_id}  # Get specific order

# Avoid verbs in URLs - use HTTP methods instead
❌ POST /api/v1/createUser
❌ GET  /api/v1/getUser?id=123
✅ POST /api/v1/users
✅ GET  /api/v1/users/123

FastAPI Implementation

from fastapi import FastAPI, HTTPException, Query, Path, Header
from pydantic import BaseModel, EmailStr
from typing import Optional, List
from datetime import datetime

app = FastAPI(title="User API", version="1.0.0")

# Request/Response models
class UserCreate(BaseModel):
    email: EmailStr
    first_name: str
    last_name: str

class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    first_name: Optional[str] = None
    last_name: Optional[str] = None

class UserResponse(BaseModel):
    id: str
    email: str
    first_name: str
    last_name: str
    created_at: datetime
    updated_at: datetime

# Endpoints
@app.get("/api/v1/users", response_model=List[UserResponse])
async def list_users(
    page: int = Query(1, ge=1, description="Page number"),
    limit: int = Query(20, ge=1, le=100, description="Items per page"),
    search: Optional[str] = Query(None, description="Search query")
):
    """List users with pagination and search."""
    skip = (page - 1) * limit
    users = db.users.find().skip(skip).limit(limit)
    return list(users)

@app.get("/api/v1/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: str = Path(..., description="User ID")
):
    """Get user by ID."""
    user = db.users.find_one({"id": user_id})
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/api/v1/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    """Create new user."""
    user_dict = user.dict()
    user_dict["id"] = generate_id()
    user_dict["created_at"] = datetime.utcnow()
    user_dict["updated_at"] = datetime.utcnow()
    
    db.users.insert_one(user_dict)
    return user_dict

@app.patch("/api/v1/users/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: str,
    updates: UserUpdate
):
    """Partially update user."""
    # Only update provided fields
    update_data = {k: v for k, v in updates.dict().items() if v is not None}
    if not update_data:
        raise HTTPException(status_code=400, detail="No fields to update")
    
    update_data["updated_at"] = datetime.utcnow()
    
    result = db.users.update_one(
        {"id": user_id},
        {"$set": update_data}
    )
    
    if result.matched_count == 0:
        raise HTTPException(status_code=404, detail="User not found")
    
    return db.users.find_one({"id": user_id})

API Versioning

# Version in URL path (my preferred method)
@app.get("/api/v1/users/{user_id}")
async def get_user_v1(user_id: str):
    """Version 1 of get user."""
    pass

@app.get("/api/v2/users/{user_id}")
async def get_user_v2(user_id: str):
    """Version 2 with additional fields."""
    pass

# Version in header
@app.get("/api/users/{user_id}")
async def get_user(
    user_id: str,
    api_version: str = Header("1.0", alias="X-API-Version")
):
    """Route based on header version."""
    if api_version == "2.0":
        return get_user_v2_logic(user_id)
    return get_user_v1_logic(user_id)

Rate Limiting

from fastapi import Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/api/v1/users")
@limiter.limit("100/minute")  # 100 requests per minute per IP
async def list_users(request: Request):
    """Rate-limited endpoint."""
    pass

# Custom rate limiting with Redis
import redis
from datetime import timedelta

redis_client = redis.Redis(host='localhost', port=6379)

class RateLimiter:
    """Custom rate limiter using Redis."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def check_rate_limit(
        self,
        key: str,
        max_requests: int,
        window_seconds: int
    ) -> tuple[bool, dict]:
        """
        Check if request is within rate limit.
        Returns (is_allowed, metadata).
        """
        pipe = self.redis.pipeline()
        now = datetime.utcnow()
        window_start = now - timedelta(seconds=window_seconds)
        
        # Remove old entries
        pipe.zremrangebyscore(key, 0, window_start.timestamp())
        # Add current request
        pipe.zadd(key, {str(now.timestamp()): now.timestamp()})
        # Count requests in window
        pipe.zcard(key)
        # Set expiry
        pipe.expire(key, window_seconds)
        
        results = pipe.execute()
        request_count = results[2]
        
        is_allowed = request_count <= max_requests
        metadata = {
            "limit": max_requests,
            "remaining": max(0, max_requests - request_count),
            "reset": int((now + timedelta(seconds=window_seconds)).timestamp())
        }
        
        return is_allowed, metadata

rate_limiter = RateLimiter(redis_client)

async def rate_limit_dependency(request: Request):
    """Dependency for rate limiting."""
    user_id = request.state.user_id  # From auth middleware
    key = f"rate_limit:{user_id}"
    
    is_allowed, metadata = rate_limiter.check_rate_limit(
        key=key,
        max_requests=1000,
        window_seconds=3600  # 1 hour
    )
    
    # Add rate limit headers
    request.state.rate_limit = metadata
    
    if not is_allowed:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded",
            headers={
                "X-RateLimit-Limit": str(metadata["limit"]),
                "X-RateLimit-Remaining": str(metadata["remaining"]),
                "X-RateLimit-Reset": str(metadata["reset"])
            }
        )

Pagination

from typing import Generic, TypeVar, List
from pydantic.generics import GenericModel

T = TypeVar('T')

class PaginatedResponse(GenericModel, Generic[T]):
    """Generic paginated response."""
    items: List[T]
    total: int
    page: int
    per_page: int
    total_pages: int

@app.get("/api/v1/users", response_model=PaginatedResponse[UserResponse])
async def list_users_paginated(
    page: int = Query(1, ge=1),
    per_page: int = Query(20, ge=1, le=100)
):
    """List users with cursor-based pagination."""
    skip = (page - 1) * per_page
    
    # Get total count
    total = db.users.count_documents({})
    
    # Get page of results
    users = list(db.users.find().skip(skip).limit(per_page))
    
    return PaginatedResponse(
        items=users,
        total=total,
        page=page,
        per_page=per_page,
        total_pages=(total + per_page - 1) // per_page
    )

# Cursor-based pagination for large datasets
@app.get("/api/v1/users/cursor")
async def list_users_cursor(
    cursor: Optional[str] = Query(None, description="Pagination cursor"),
    limit: int = Query(20, ge=1, le=100)
):
    """Cursor-based pagination for efficient large dataset traversal."""
    query = {}
    
    if cursor:
        # Decode cursor (base64 encoded timestamp or ID)
        cursor_data = decode_cursor(cursor)
        query["created_at"] = {"$gt": cursor_data["created_at"]}
    
    users = list(
        db.users.find(query)
        .sort("created_at", 1)
        .limit(limit + 1)  # Fetch one extra to determine if there's more
    )
    
    has_more = len(users) > limit
    if has_more:
        users = users[:limit]
    
    next_cursor = None
    if has_more and users:
        # Create cursor from last item
        next_cursor = encode_cursor({
            "created_at": users[-1]["created_at"]
        })
    
    return {
        "items": users,
        "next_cursor": next_cursor,
        "has_more": has_more
    }

Idempotency

from fastapi import Header

class IdempotencyManager:
    """Manage idempotent requests using Redis."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 86400  # 24 hours
    
    def check_idempotency(self, idempotency_key: str) -> Optional[dict]:
        """Check if request was already processed."""
        cached = self.redis.get(f"idempotency:{idempotency_key}")
        if cached:
            return json.loads(cached)
        return None
    
    def store_result(self, idempotency_key: str, result: dict):
        """Store result for future duplicate requests."""
        self.redis.setex(
            f"idempotency:{idempotency_key}",
            self.ttl,
            json.dumps(result)
        )

idempotency_mgr = IdempotencyManager(redis_client)

@app.post("/api/v1/payments")
async def create_payment(
    payment_data: PaymentCreate,
    idempotency_key: str = Header(..., alias="Idempotency-Key")
):
    """
    Idempotent payment creation.
    Same idempotency key returns same result.
    """
    # Check if already processed
    existing = idempotency_mgr.check_idempotency(idempotency_key)
    if existing:
        return existing
    
    # Process payment
    payment = process_payment(payment_data)
    
    # Store result
    result = {"payment_id": payment["id"], "status": "success"}
    idempotency_mgr.store_result(idempotency_key, result)
    
    return result

GraphQL API

GraphQL solves over-fetching and under-fetching problems in REST.

import strawberry
from typing import List, Optional

@strawberry.type
class User:
    id: str
    email: str
    first_name: str
    last_name: str
    
    @strawberry.field
    def full_name(self) -> str:
        return f"{self.first_name} {self.last_name}"
    
    @strawberry.field
    async def orders(self) -> List['Order']:
        """Lazy load user's orders."""
        return await get_user_orders(self.id)

@strawberry.type
class Order:
    id: str
    total: float
    status: str
    created_at: str

@strawberry.type
class Query:
    @strawberry.field
    async def user(self, id: str) -> Optional[User]:
        """Get user by ID."""
        user_data = db.users.find_one({"id": id})
        return User(**user_data) if user_data else None
    
    @strawberry.field
    async def users(
        self,
        page: int = 1,
        limit: int = 20
    ) -> List[User]:
        """List users."""
        skip = (page - 1) * limit
        users_data = db.users.find().skip(skip).limit(limit)
        return [User(**u) for u in users_data]

@strawberry.type
class Mutation:
    @strawberry.mutation
    async def create_user(
        self,
        email: str,
        first_name: str,
        last_name: str
    ) -> User:
        """Create new user."""
        user_data = {
            "id": generate_id(),
            "email": email,
            "first_name": first_name,
            "last_name": last_name
        }
        db.users.insert_one(user_data)
        return User(**user_data)

# Create schema
schema = strawberry.Schema(query=Query, mutation=Mutation)

# Add to FastAPI
from strawberry.fastapi import GraphQLRouter

graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

gRPC API

gRPC for high-performance internal service communication.

// users.proto
syntax = "proto3";

package users;

service UserService {
  rpc GetUser (GetUserRequest) returns (User);
  rpc ListUsers (ListUsersRequest) returns (ListUsersResponse);
  rpc CreateUser (CreateUserRequest) returns (User);
}

message User {
  string id = 1;
  string email = 2;
  string first_name = 3;
  string last_name = 4;
}

message GetUserRequest {
  string id = 1;
}

message ListUsersRequest {
  int32 page = 1;
  int32 limit = 2;
}

message ListUsersResponse {
  repeated User users = 1;
  int32 total = 2;
}

message CreateUserRequest {
  string email = 1;
  string first_name = 2;
  string last_name = 3;
}

# Python gRPC server implementation
import grpc
from concurrent import futures
import users_pb2
import users_pb2_grpc

class UserServicer(users_pb2_grpc.UserServiceServicer):
    """gRPC UserService implementation."""
    
    def GetUser(self, request, context):
        """Get user by ID."""
        user_data = db.users.find_one({"id": request.id})
        
        if not user_data:
            context.abort(grpc.StatusCode.NOT_FOUND, "User not found")
        
        return users_pb2.User(
            id=user_data["id"],
            email=user_data["email"],
            first_name=user_data["first_name"],
            last_name=user_data["last_name"]
        )
    
    def ListUsers(self, request, context):
        """List users with pagination."""
        skip = (request.page - 1) * request.limit
        users_data = db.users.find().skip(skip).limit(request.limit)
        total = db.users.count_documents({})
        
        users = [
            users_pb2.User(
                id=u["id"],
                email=u["email"],
                first_name=u["first_name"],
                last_name=u["last_name"]
            )
            for u in users_data
        ]
        
        return users_pb2.ListUsersResponse(users=users, total=total)

# Start server
def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    users_pb2_grpc.add_UserServiceServicer_to_server(UserServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

Error Handling

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

class APIError(Exception):
    """Base API error."""
    def __init__(self, status_code: int, message: str, details: dict = None):
        self.status_code = status_code
        self.message = message
        self.details = details or {}

# Global exception handlers
@app.exception_handler(APIError)
async def api_error_handler(request: Request, exc: APIError):
    """Handle custom API errors."""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "message": exc.message,
                "details": exc.details,
                "path": str(request.url),
                "timestamp": datetime.utcnow().isoformat()
            }
        }
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Handle validation errors."""
    return JSONResponse(
        status_code=422,
        content={
            "error": {
                "message": "Validation error",
                "details": exc.errors(),
                "path": str(request.url)
            }
        }
    )

# Usage
@app.get("/api/v1/users/{user_id}")
async def get_user(user_id: str):
    user = db.users.find_one({"id": user_id})
    if not user:
        raise APIError(
            status_code=404,
            message="User not found",
            details={"user_id": user_id}
        )
    return user

API Documentation

from fastapi.openapi.utils import get_openapi

def custom_openapi():
    """Custom OpenAPI schema with additional metadata."""
    if app.openapi_schema:
        return app.openapi_schema
    
    openapi_schema = get_openapi(
        title="User API",
        version="1.0.0",
        description="User management API with full CRUD operations",
        routes=app.routes,
    )
    
    # Add security schemes
    openapi_schema["components"]["securitySchemes"] = {
        "bearerAuth": {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "JWT"
        }
    }
    
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

API Testing

from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_create_user():
    """Test user creation."""
    response = client.post(
        "/api/v1/users",
        json={
            "email": "[email protected]",
            "first_name": "Test",
            "last_name": "User"
        }
    )
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "[email protected]"
    assert "id" in data

def test_get_user_not_found():
    """Test getting non-existent user."""
    response = client.get("/api/v1/users/nonexistent")
    assert response.status_code == 404

Lessons Learned

What worked:

1. REST for public APIs - widely understood

2. GraphQL for complex data requirements

3. gRPC for internal service communication

4. Versioning from day one

5. Comprehensive error responses

What didn't work:

1. No API versioning strategy

2. Inconsistent error responses

3. Missing rate limiting

4. Poor documentation

5. Not implementing idempotency for critical operations

What's Next

Understanding API design, let's explore microservices patterns:

• Microservices Patterns →: Service decomposition and communication

---

Navigation:

• ← Previous: Message Queues

• Next: Microservices Patterns →

• Back to Overview