Intermediate

Build REST APIs

REST APIs are the backbone of every web service. In systems coding interviews, you may be asked to build an in-memory CRUD API, implement pagination, add authentication middleware, or design structured error responses. This lesson covers four complete implementations you can adapt to any interview scenario.

Problem 1: In-Memory CRUD API

Build a complete REST-style API that supports Create, Read, Update, and Delete operations on an in-memory data store. This is the foundation for every other problem in this lesson.

import uuid
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from enum import Enum


class HttpStatus(Enum):
    OK = 200
    CREATED = 201
    NO_CONTENT = 204
    BAD_REQUEST = 400
    NOT_FOUND = 404
    CONFLICT = 409
    INTERNAL_ERROR = 500


@dataclass
class ApiResponse:
    status: int
    body: Any = None
    headers: Dict[str, str] = field(default_factory=dict)


class InMemoryCrudApi:
    """RESTful CRUD API backed by an in-memory dictionary.

    Supports:
    - POST /resources      -> create a new resource
    - GET /resources/:id   -> read a single resource
    - GET /resources       -> list all resources
    - PUT /resources/:id   -> full update of a resource
    - PATCH /resources/:id -> partial update of a resource
    - DELETE /resources/:id -> delete a resource
    """

    def __init__(self):
        self._store: Dict[str, Dict] = {}

    def create(self, data: Dict) -> ApiResponse:
        """POST /resources - Create a new resource."""
        if not data:
            return ApiResponse(
                status=HttpStatus.BAD_REQUEST.value,
                body={"error": "Request body cannot be empty"}
            )

        resource_id = str(uuid.uuid4())
        now = time.time()

        resource = {
            "id": resource_id,
            **data,
            "created_at": now,
            "updated_at": now,
        }

        self._store[resource_id] = resource

        return ApiResponse(
            status=HttpStatus.CREATED.value,
            body=resource,
            headers={"Location": f"/resources/{resource_id}"}
        )

    def get(self, resource_id: str) -> ApiResponse:
        """GET /resources/:id - Read a single resource."""
        resource = self._store.get(resource_id)
        if resource is None:
            return ApiResponse(
                status=HttpStatus.NOT_FOUND.value,
                body={"error": f"Resource {resource_id} not found"}
            )
        return ApiResponse(status=HttpStatus.OK.value, body=resource)

    def list_all(self) -> ApiResponse:
        """GET /resources - List all resources."""
        resources = list(self._store.values())
        return ApiResponse(
            status=HttpStatus.OK.value,
            body={"data": resources, "total": len(resources)}
        )

    def update(self, resource_id: str, data: Dict) -> ApiResponse:
        """PUT /resources/:id - Full replacement update."""
        if resource_id not in self._store:
            return ApiResponse(
                status=HttpStatus.NOT_FOUND.value,
                body={"error": f"Resource {resource_id} not found"}
            )

        existing = self._store[resource_id]
        updated = {
            "id": resource_id,
            **data,
            "created_at": existing["created_at"],
            "updated_at": time.time(),
        }
        self._store[resource_id] = updated

        return ApiResponse(status=HttpStatus.OK.value, body=updated)

    def partial_update(self, resource_id: str, data: Dict) -> ApiResponse:
        """PATCH /resources/:id - Partial update."""
        if resource_id not in self._store:
            return ApiResponse(
                status=HttpStatus.NOT_FOUND.value,
                body={"error": f"Resource {resource_id} not found"}
            )

        resource = self._store[resource_id]
        resource.update(data)
        resource["updated_at"] = time.time()

        return ApiResponse(status=HttpStatus.OK.value, body=resource)

    def delete(self, resource_id: str) -> ApiResponse:
        """DELETE /resources/:id - Delete a resource."""
        if resource_id not in self._store:
            return ApiResponse(
                status=HttpStatus.NOT_FOUND.value,
                body={"error": f"Resource {resource_id} not found"}
            )

        del self._store[resource_id]
        return ApiResponse(status=HttpStatus.NO_CONTENT.value)


# ---- Usage ----
api = InMemoryCrudApi()

# Create
resp = api.create({"name": "Alice", "email": "alice@example.com"})
print(f"CREATE: {resp.status} -> {resp.body['id']}")

# Read
user_id = resp.body["id"]
resp = api.get(user_id)
print(f"GET: {resp.status} -> {resp.body['name']}")

# Update
resp = api.partial_update(user_id, {"name": "Alice Smith"})
print(f"PATCH: {resp.status} -> {resp.body['name']}")

# List
resp = api.list_all()
print(f"LIST: {resp.status} -> {resp.body['total']} resources")

# Delete
resp = api.delete(user_id)
print(f"DELETE: {resp.status}")

# Verify deletion
resp = api.get(user_id)
print(f"GET after delete: {resp.status}")  # 404

Problem 2: Pagination

Implement both offset-based and cursor-based pagination. Understand the trade-offs: offset pagination is simple but breaks when data changes; cursor pagination is stable but harder to implement.

class PaginatedApi:
    """API with both offset and cursor-based pagination."""

    def __init__(self):
        self._store: Dict[str, Dict] = {}
        self._ordered_ids: List[str] = []  # Insertion order

    def create(self, data: Dict) -> ApiResponse:
        resource_id = str(uuid.uuid4())
        resource = {"id": resource_id, **data, "created_at": time.time()}
        self._store[resource_id] = resource
        self._ordered_ids.append(resource_id)
        return ApiResponse(status=201, body=resource)

    def list_offset(self, page: int = 1, page_size: int = 10) -> ApiResponse:
        """Offset-based pagination: GET /resources?page=2&page_size=10

        Pros: Simple, supports jumping to any page
        Cons: Breaks if items are inserted/deleted between pages
        """
        if page < 1 or page_size < 1:
            return ApiResponse(status=400, body={"error": "Invalid pagination"})

        total = len(self._ordered_ids)
        start = (page - 1) * page_size
        end = start + page_size

        ids = self._ordered_ids[start:end]
        items = [self._store[rid] for rid in ids]

        return ApiResponse(
            status=200,
            body={
                "data": items,
                "pagination": {
                    "page": page,
                    "page_size": page_size,
                    "total": total,
                    "total_pages": (total + page_size - 1) // page_size,
                    "has_next": end < total,
                    "has_prev": page > 1,
                }
            }
        )

    def list_cursor(self, cursor: Optional[str] = None,
                    limit: int = 10) -> ApiResponse:
        """Cursor-based pagination: GET /resources?cursor=abc&limit=10

        Pros: Stable under insertions/deletions, O(1) seek
        Cons: Cannot jump to arbitrary page, cursor must be opaque
        """
        if limit < 1:
            return ApiResponse(status=400, body={"error": "Invalid limit"})

        if cursor is None:
            start_idx = 0
        else:
            # Find position after cursor
            try:
                start_idx = self._ordered_ids.index(cursor) + 1
            except ValueError:
                return ApiResponse(status=400, body={"error": "Invalid cursor"})

        end_idx = start_idx + limit
        ids = self._ordered_ids[start_idx:end_idx]
        items = [self._store[rid] for rid in ids]

        next_cursor = ids[-1] if len(ids) == limit and end_idx < len(self._ordered_ids) else None

        return ApiResponse(
            status=200,
            body={
                "data": items,
                "pagination": {
                    "next_cursor": next_cursor,
                    "has_more": next_cursor is not None,
                    "count": len(items),
                }
            }
        )


# ---- Usage ----
api = PaginatedApi()
for i in range(25):
    api.create({"name": f"User {i}"})

# Offset pagination
resp = api.list_offset(page=1, page_size=10)
print(f"Page 1: {len(resp.body['data'])} items, "
      f"total_pages: {resp.body['pagination']['total_pages']}")

# Cursor pagination
resp = api.list_cursor(limit=10)
print(f"First batch: {len(resp.body['data'])} items")
next_cursor = resp.body['pagination']['next_cursor']

resp = api.list_cursor(cursor=next_cursor, limit=10)
print(f"Second batch: {len(resp.body['data'])} items")

Problem 3: Authentication Middleware

Implement a JWT-style authentication middleware that validates tokens and extracts user identity. In interviews, you typically implement the middleware pattern rather than full JWT cryptography.

import hashlib
import json
import base64


class AuthMiddleware:
    """Simple JWT-style authentication middleware.

    In a real system, use a proper JWT library. For interviews,
    this demonstrates the middleware pattern and token validation.
    """

    def __init__(self, secret_key: str):
        self._secret = secret_key
        self._revoked_tokens: set = set()

    def create_token(self, user_id: str, roles: List[str],
                     expires_in: int = 3600) -> str:
        """Create a signed token (simplified JWT)."""
        header = {"alg": "HS256", "typ": "JWT"}
        payload = {
            "sub": user_id,
            "roles": roles,
            "iat": time.time(),
            "exp": time.time() + expires_in,
        }

        header_b64 = base64.urlsafe_b64encode(
            json.dumps(header).encode()
        ).decode().rstrip("=")

        payload_b64 = base64.urlsafe_b64encode(
            json.dumps(payload).encode()
        ).decode().rstrip("=")

        message = f"{header_b64}.{payload_b64}"
        signature = hashlib.sha256(
            f"{message}.{self._secret}".encode()
        ).hexdigest()[:32]

        return f"{message}.{signature}"

    def validate_token(self, token: str) -> ApiResponse:
        """Validate a token and return the decoded payload."""
        if token in self._revoked_tokens:
            return ApiResponse(status=401, body={"error": "Token revoked"})

        parts = token.split(".")
        if len(parts) != 3:
            return ApiResponse(status=401, body={"error": "Malformed token"})

        header_b64, payload_b64, signature = parts

        # Verify signature
        message = f"{header_b64}.{payload_b64}"
        expected_sig = hashlib.sha256(
            f"{message}.{self._secret}".encode()
        ).hexdigest()[:32]

        if signature != expected_sig:
            return ApiResponse(status=401, body={"error": "Invalid signature"})

        # Decode payload
        padding = 4 - len(payload_b64) % 4
        payload_b64 += "=" * padding
        payload = json.loads(base64.urlsafe_b64decode(payload_b64))

        # Check expiration
        if payload.get("exp", 0) < time.time():
            return ApiResponse(status=401, body={"error": "Token expired"})

        return ApiResponse(status=200, body=payload)

    def revoke_token(self, token: str):
        """Add token to revocation list (logout)."""
        self._revoked_tokens.add(token)

    def require_auth(self, handler):
        """Middleware decorator: authenticate before executing handler."""
        def wrapped(request: Dict) -> ApiResponse:
            token = request.get("headers", {}).get("Authorization", "")
            if token.startswith("Bearer "):
                token = token[7:]

            auth_result = self.validate_token(token)
            if auth_result.status != 200:
                return auth_result

            # Inject user info into request
            request["user"] = auth_result.body
            return handler(request)
        return wrapped

    def require_role(self, role: str):
        """Middleware decorator: require a specific role."""
        def decorator(handler):
            def wrapped(request: Dict) -> ApiResponse:
                # First authenticate
                token = request.get("headers", {}).get("Authorization", "")
                if token.startswith("Bearer "):
                    token = token[7:]

                auth_result = self.validate_token(token)
                if auth_result.status != 200:
                    return auth_result

                # Check role
                user_roles = auth_result.body.get("roles", [])
                if role not in user_roles:
                    return ApiResponse(
                        status=403,
                        body={"error": f"Requires role: {role}"}
                    )

                request["user"] = auth_result.body
                return handler(request)
            return wrapped
        return decorator


# ---- Usage ----
auth = AuthMiddleware(secret_key="my-secret-key-123")

# Create token
token = auth.create_token("user_42", roles=["user", "admin"])
print(f"Token: {token[:50]}...")

# Validate token
result = auth.validate_token(token)
print(f"Valid: {result.status == 200}, User: {result.body['sub']}")

# Middleware pattern
@auth.require_auth
def get_profile(request):
    return ApiResponse(status=200, body={"user": request["user"]["sub"]})

resp = get_profile({"headers": {"Authorization": f"Bearer {token}"}})
print(f"Profile: {resp.body}")

# Role-based access
@auth.require_role("admin")
def delete_user(request):
    return ApiResponse(status=200, body={"deleted": True})

resp = delete_user({"headers": {"Authorization": f"Bearer {token}"}})
print(f"Admin action: {resp.status}")  # 200 (has admin role)

Problem 4: Structured Error Handling

Implement a centralized error handling system that produces consistent, informative error responses. This is critical for production APIs and frequently tested in interviews.

from dataclasses import dataclass
from typing import Optional, List, Dict
from enum import Enum


class ErrorCode(Enum):
    VALIDATION_ERROR = "VALIDATION_ERROR"
    NOT_FOUND = "NOT_FOUND"
    UNAUTHORIZED = "UNAUTHORIZED"
    FORBIDDEN = "FORBIDDEN"
    CONFLICT = "CONFLICT"
    RATE_LIMITED = "RATE_LIMITED"
    INTERNAL_ERROR = "INTERNAL_ERROR"


@dataclass
class ApiError:
    code: ErrorCode
    message: str
    details: Optional[List[Dict]] = None
    status: int = 400

    def to_response(self) -> ApiResponse:
        body = {
            "error": {
                "code": self.code.value,
                "message": self.message,
            }
        }
        if self.details:
            body["error"]["details"] = self.details
        return ApiResponse(status=self.status, body=body)


class ErrorHandler:
    """Centralized error handling for API endpoints."""

    # Map error codes to HTTP status codes
    STATUS_MAP = {
        ErrorCode.VALIDATION_ERROR: 400,
        ErrorCode.NOT_FOUND: 404,
        ErrorCode.UNAUTHORIZED: 401,
        ErrorCode.FORBIDDEN: 403,
        ErrorCode.CONFLICT: 409,
        ErrorCode.RATE_LIMITED: 429,
        ErrorCode.INTERNAL_ERROR: 500,
    }

    @classmethod
    def validation_error(cls, field: str, message: str) -> ApiResponse:
        return ApiError(
            code=ErrorCode.VALIDATION_ERROR,
            message="Validation failed",
            details=[{"field": field, "message": message}],
            status=400,
        ).to_response()

    @classmethod
    def not_found(cls, resource: str, resource_id: str) -> ApiResponse:
        return ApiError(
            code=ErrorCode.NOT_FOUND,
            message=f"{resource} with id '{resource_id}' not found",
            status=404,
        ).to_response()

    @classmethod
    def validate_request(cls, data: Dict,
                         required_fields: List[str]) -> Optional[ApiResponse]:
        """Validate that all required fields are present and non-empty."""
        errors = []
        for field_name in required_fields:
            if field_name not in data:
                errors.append({
                    "field": field_name,
                    "message": f"'{field_name}' is required"
                })
            elif not data[field_name]:
                errors.append({
                    "field": field_name,
                    "message": f"'{field_name}' cannot be empty"
                })

        if errors:
            return ApiError(
                code=ErrorCode.VALIDATION_ERROR,
                message="Validation failed",
                details=errors,
                status=400,
            ).to_response()

        return None  # Validation passed

    @classmethod
    def handle_exception(cls, handler):
        """Decorator that catches exceptions and returns error responses."""
        def wrapped(*args, **kwargs):
            try:
                return handler(*args, **kwargs)
            except KeyError as e:
                return ApiError(
                    code=ErrorCode.NOT_FOUND,
                    message=f"Resource not found: {e}",
                    status=404,
                ).to_response()
            except ValueError as e:
                return ApiError(
                    code=ErrorCode.VALIDATION_ERROR,
                    message=str(e),
                    status=400,
                ).to_response()
            except Exception as e:
                return ApiError(
                    code=ErrorCode.INTERNAL_ERROR,
                    message="An unexpected error occurred",
                    status=500,
                ).to_response()
        return wrapped


# ---- Usage: Putting it all together ----
class UserApi:
    def __init__(self):
        self._store = {}

    @ErrorHandler.handle_exception
    def create_user(self, data: Dict) -> ApiResponse:
        # Validate required fields
        error = ErrorHandler.validate_request(data, ["name", "email"])
        if error:
            return error

        # Check for duplicate email
        for user in self._store.values():
            if user["email"] == data["email"]:
                return ApiError(
                    code=ErrorCode.CONFLICT,
                    message=f"User with email '{data['email']}' already exists",
                    status=409,
                ).to_response()

        user_id = str(uuid.uuid4())
        user = {"id": user_id, **data}
        self._store[user_id] = user
        return ApiResponse(status=201, body=user)


user_api = UserApi()

# Missing field
resp = user_api.create_user({"name": "Alice"})
print(f"Missing email: {resp.status} -> {resp.body}")

# Success
resp = user_api.create_user({"name": "Alice", "email": "alice@test.com"})
print(f"Created: {resp.status} -> {resp.body['id']}")

# Duplicate
resp = user_api.create_user({"name": "Bob", "email": "alice@test.com"})
print(f"Duplicate: {resp.status} -> {resp.body}")

💡

Interview tip: When asked to build a REST API, always include validation and error handling even if not explicitly asked. Returning structured error responses with appropriate HTTP status codes signals production experience. Use enums for error codes to show you think about consistency.

Key Takeaways

💡

CRUD APIs need proper status codes: 201 for create, 204 for delete, 404 for not found
Offset pagination is simple but breaks under mutations; cursor pagination is stable
Authentication middleware follows the decorator pattern: validate token, inject user, call handler
Structured error responses need a consistent format: error code, message, and optional details
Always validate inputs and handle edge cases (empty body, missing fields, duplicates)
Use dataclasses and enums for clean, type-safe code that impresses interviewers

← Previous System Implementation in Interviews Next → Implement Rate Limiters