System Design Series Part 6: API Design & REST/GraphQL

# HTTP Status Codes - Use them correctly!

# 2xx Success
200 OK              # General success, return data
201 Created         # Resource created (return location header)
204 No Content      # Success, no response body (DELETE)

# 3xx Redirection
301 Moved Permanently    # Resource URL changed
304 Not Modified         # Cached version is current

# 4xx Client Errors
400 Bad Request     # Invalid syntax, validation error
401 Unauthorized    # Authentication required/failed
403 Forbidden       # Authenticated but not authorized
404 Not Found       # Resource doesn't exist
405 Method Not Allowed   # HTTP method not supported
409 Conflict        # State conflict (duplicate, version)
422 Unprocessable Entity  # Valid syntax, semantic error
429 Too Many Requests    # Rate limit exceeded

# 5xx Server Errors
500 Internal Server Error  # Unexpected server error
502 Bad Gateway            # Upstream server error
503 Service Unavailable    # Maintenance or overload
504 Gateway Timeout        # Upstream timeout

# GraphQL Schema (SDL)
type User {
    id: ID!
    name: String!
    email: String!
    posts: [Post!]!
    followers: [User!]!
    createdAt: DateTime!
}

type Post {
    id: ID!
    title: String!
    content: String!
    author: User!
    comments: [Comment!]!
    likes: Int!
    publishedAt: DateTime
}

type Comment {
    id: ID!
    text: String!
    author: User!
    createdAt: DateTime!
}

type Query {
    user(id: ID!): User
    users(limit: Int, offset: Int): [User!]!
    post(id: ID!): Post
    posts(authorId: ID, published: Boolean): [Post!]!
}

type Mutation {
    createUser(input: CreateUserInput!): User!
    updateUser(id: ID!, input: UpdateUserInput!): User!
    createPost(input: CreatePostInput!): Post!
    deletePost(id: ID!): Boolean!
}

input CreateUserInput {
    name: String!
    email: String!
}

input CreatePostInput {
    title: String!
    content: String!
}

# GraphQL Query - Client requests exactly what it needs
query GetUserWithPosts($userId: ID!) {
    user(id: $userId) {
        name
        email
        posts {
            id
            title
            likes
            comments {
                text
                author {
                    name
                }
            }
        }
    }
}

# Variables
{
    "userId": "123"
}

# Response - Only requested fields
{
    "data": {
        "user": {
            "name": "John Doe",
            "email": "john@example.com",
            "posts": [
                {
                    "id": "post_1",
                    "title": "My First Post",
                    "likes": 42,
                    "comments": [
                        {
                            "text": "Great post!",
                            "author": { "name": "Jane Smith" }
                        }
                    ]
                }
            ]
        }
    }
}

# GraphQL Mutation - Create/Update data
mutation CreatePost($input: CreatePostInput!) {
    createPost(input: $input) {
        id
        title
        content
        publishedAt
        author {
            name
        }
    }
}

# Variables
{
    "input": {
        "title": "GraphQL Best Practices",
        "content": "Here's what I learned..."
    }
}

# GraphQL Subscription - Real-time updates
subscription OnNewComment($postId: ID!) {
    commentAdded(postId: $postId) {
        id
        text
        author {
            name
            avatar
        }
        createdAt
    }
}

# Server pushes updates when new comments arrive

# JWT Authentication Flow
import jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"

def create_token(user_id):
    payload = {
        "sub": user_id,
        "iat": datetime.utcnow(),
        "exp": datetime.utcnow() + timedelta(hours=24)
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def verify_token(token):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload["sub"]
    except jwt.ExpiredSignatureError:
        raise APIError("TOKEN_EXPIRED", "Token has expired", status_code=401)
    except jwt.InvalidTokenError:
        raise APIError("INVALID_TOKEN", "Invalid token", status_code=401)

# Usage in requests
# Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

For third-party authentication and authorization:

# OAuth 2.0 Authorization Code Flow
# 1. Redirect user to authorization server
GET https://auth.example.com/authorize?
    response_type=code&
    client_id=YOUR_CLIENT_ID&
    redirect_uri=https://yourapp.com/callback&
    scope=openid profile email&
    state=random_state

# 2. User authenticates and grants permission

# 3. Receive authorization code at callback
GET https://yourapp.com/callback?code=AUTH_CODE&state=random_state

# 4. Exchange code for tokens
POST https://auth.example.com/token
Content-Type: application/x-www-form-urlencoded

grant_type=authorization_code&
code=AUTH_CODE&
redirect_uri=https://yourapp.com/callback&
client_id=YOUR_CLIENT_ID&
client_secret=YOUR_CLIENT_SECRET

# 5. Receive access token and ID token
{
    "access_token": "eyJ...",
    "token_type": "Bearer",
    "expires_in": 3600,
    "refresh_token": "dGhpc...",
    "id_token": "eyJ..."
}

For server-to-server authentication:

# API Key Authentication
# Header approach (recommended)
X-API-Key: your-api-key-here

# Implementation
from functools import wraps
from flask import request

def require_api_key(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        api_key = request.headers.get('X-API-Key')
        
        if not api_key:
            raise APIError("MISSING_API_KEY", "API key required", status_code=401)
        
        # Validate key (check database, cache)
        client = validate_api_key(api_key)
        if not client:
            raise APIError("INVALID_API_KEY", "Invalid API key", status_code=401)
        
        # Attach client info for rate limiting, logging
        request.client = client
        return f(*args, **kwargs)
    
    return decorated

# Server-side (Python with websockets library)
import asyncio
import websockets
import json

connected_clients = set()

async def handler(websocket, path):
    # Register client
    connected_clients.add(websocket)
    
    try:
        async for message in websocket:
            data = json.loads(message)
            
            if data["type"] == "chat_message":
                # Broadcast to all connected clients
                await broadcast({
                    "type": "new_message",
                    "user": data["user"],
                    "message": data["message"],
                    "timestamp": datetime.utcnow().isoformat()
                })
                
    finally:
        connected_clients.remove(websocket)

async def broadcast(message):
    if connected_clients:
        await asyncio.gather(
            *[client.send(json.dumps(message)) for client in connected_clients]
        )

# Start server
asyncio.run(websockets.serve(handler, "localhost", 8765))

// Client-side (JavaScript)
const socket = new WebSocket('wss://example.com/ws');

socket.onopen = () => {
    console.log('Connected');
    socket.send(JSON.stringify({
        type: 'chat_message',
        user: 'John',
        message: 'Hello!'
    }));
};

socket.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log('Received:', data);
};

socket.onclose = () => console.log('Disconnected');
socket.onerror = (error) => console.error('Error:', error);

# user.proto - Define service and messages
syntax = "proto3";

package user;

service UserService {
    // Unary RPC
    rpc GetUser(GetUserRequest) returns (User);
    
    // Server streaming
    rpc ListUsers(ListUsersRequest) returns (stream User);
    
    // Client streaming
    rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);
    
    // Bidirectional streaming
    rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message User {
    string id = 1;
    string name = 2;
    string email = 3;
    int32 age = 4;
}

message GetUserRequest {
    string id = 1;
}

message ListUsersRequest {
    int32 page_size = 1;
    string page_token = 2;
}

# Python gRPC Server
import grpc
from concurrent import futures
import user_pb2
import user_pb2_grpc

class UserService(user_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        # Fetch user from database
        user = db.get_user(request.id)
        
        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f"User {request.id} not found")
            return user_pb2.User()
        
        return user_pb2.User(
            id=user.id,
            name=user.name,
            email=user.email,
            age=user.age
        )
    
    def ListUsers(self, request, context):
        # Server streaming - yield multiple responses
        users = db.list_users(page_size=request.page_size)
        for user in users:
            yield user_pb2.User(
                id=user.id,
                name=user.name,
                email=user.email
            )

# Start server
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
user_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
server.add_insecure_port('[::]:50051')
server.start()
server.wait_for_termination()