API Design for Microservices

Introduction

APIs are the contracts between your services. Through designing APIs for systems handling millions of requests, I've learned that good API design isn't just about following REST conventions—it's about creating intuitive, evolvable interfaces that teams can depend on for years.

This article covers practical API design principles, versioning strategies, and how to use OpenAPI with FastAPI for contract-first development.

API Design Principles

REST Fundamentals

REST (Representational State Transfer) provides a consistent way to design APIs around resources.

Resource Naming

# ✅ Good: Nouns for resources, plural forms
GET    /users              # List users
GET    /users/{id}         # Get specific user
POST   /users              # Create user
PUT    /users/{id}         # Replace user
PATCH  /users/{id}         # Partial update
DELETE /users/{id}         # Delete user

# ✅ Good: Nested resources for relationships
GET    /users/{id}/orders           # User's orders
GET    /orders/{id}/items           # Order's items
POST   /users/{id}/addresses        # Add address to user

# ❌ Bad: Verbs in URLs (RPC-style)
POST   /getUser
POST   /createUser
POST   /deleteUser
GET    /fetchAllOrders

HTTP Methods and Status Codes

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()


class UserCreate(BaseModel):
    email: str
    name: str


class User(BaseModel):
    id: str
    email: str
    name: str


# GET - Retrieve resource (idempotent, cacheable)
@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: str):
    user = await user_repository.find_by_id(user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User {user_id} not found"
        )
    return user


# POST - Create resource (not idempotent)
@app.post("/users", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(user_data: UserCreate):
    user = await user_repository.create(user_data)
    return user


# PUT - Replace resource (idempotent)
@app.put("/users/{user_id}", response_model=User)
async def replace_user(user_id: str, user_data: UserCreate):
    user = await user_repository.replace(user_id, user_data)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


# PATCH - Partial update (may not be idempotent)
class UserUpdate(BaseModel):
    email: str | None = None
    name: str | None = None


@app.patch("/users/{user_id}", response_model=User)
async def update_user(user_id: str, user_data: UserUpdate):
    updates = user_data.model_dump(exclude_unset=True)
    user = await user_repository.update(user_id, updates)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


# DELETE - Remove resource (idempotent)
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: str):
    deleted = await user_repository.delete(user_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="User not found")

Status Code Reference

Code	Meaning	When to Use
200	OK	Successful GET, PUT, PATCH
201	Created	Successful POST
204	No Content	Successful DELETE
400	Bad Request	Invalid input
401	Unauthorized	Missing/invalid authentication
403	Forbidden	Valid auth but insufficient permissions
404	Not Found	Resource doesn't exist
409	Conflict	Resource state conflict
422	Unprocessable Entity	Validation error
500	Internal Server Error	Unexpected server error
503	Service Unavailable	Temporary unavailability

Request and Response Design

Consistent Response Format

from pydantic import BaseModel
from typing import Generic, TypeVar

T = TypeVar("T")


class PaginationMeta(BaseModel):
    page: int
    page_size: int
    total_items: int
    total_pages: int


class ListResponse(BaseModel, Generic[T]):
    """Consistent list response with pagination."""
    data: list[T]
    meta: PaginationMeta


class ErrorDetail(BaseModel):
    field: str | None = None
    message: str
    code: str


class ErrorResponse(BaseModel):
    """Consistent error response format."""
    error: str
    message: str
    details: list[ErrorDetail] = []
    request_id: str


# Usage in FastAPI
@app.get("/users", response_model=ListResponse[User])
async def list_users(page: int = 1, page_size: int = 20):
    users, total = await user_repository.find_paginated(page, page_size)
    return ListResponse(
        data=users,
        meta=PaginationMeta(
            page=page,
            page_size=page_size,
            total_items=total,
            total_pages=(total + page_size - 1) // page_size,
        ),
    )

Error Handling

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import uuid


# Custom exception classes
class DomainException(Exception):
    def __init__(self, message: str, code: str = "DOMAIN_ERROR"):
        self.message = message
        self.code = code


class ResourceNotFoundError(DomainException):
    def __init__(self, resource: str, resource_id: str):
        super().__init__(
            message=f"{resource} with id '{resource_id}' not found",
            code="RESOURCE_NOT_FOUND"
        )
        self.resource = resource
        self.resource_id = resource_id


class BusinessRuleViolationError(DomainException):
    def __init__(self, message: str):
        super().__init__(message=message, code="BUSINESS_RULE_VIOLATION")


# Exception handlers
@app.exception_handler(ResourceNotFoundError)
async def resource_not_found_handler(request: Request, exc: ResourceNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "error": "not_found",
            "message": exc.message,
            "details": [],
            "request_id": request.state.request_id,
        },
    )


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    details = [
        {
            "field": ".".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "code": error["type"],
        }
        for error in exc.errors()
    ]
    return JSONResponse(
        status_code=422,
        content={
            "error": "validation_error",
            "message": "Request validation failed",
            "details": details,
            "request_id": request.state.request_id,
        },
    )


# Middleware to add request ID
@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    request.state.request_id = request_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

API Versioning Strategies

URL Path Versioning

from fastapi import APIRouter

# Version 1
v1_router = APIRouter(prefix="/api/v1")

@v1_router.get("/users/{user_id}")
async def get_user_v1(user_id: str):
    """Original user endpoint."""
    user = await user_repository.find_by_id(user_id)
    return {
        "id": user.id,
        "email": user.email,
        "name": user.name,  # Original: single name field
    }


# Version 2 - Breaking change: split name into first/last
v2_router = APIRouter(prefix="/api/v2")

@v2_router.get("/users/{user_id}")
async def get_user_v2(user_id: str):
    """Updated user endpoint with first/last name."""
    user = await user_repository.find_by_id(user_id)
    return {
        "id": user.id,
        "email": user.email,
        "first_name": user.first_name,  # New: split name
        "last_name": user.last_name,
    }


# Mount both versions
app.include_router(v1_router)
app.include_router(v2_router)

Header-Based Versioning

from fastapi import Header, Depends


async def get_api_version(
    accept: str = Header(default="application/vnd.api.v1+json")
) -> int:
    """Extract API version from Accept header."""
    if "v2" in accept:
        return 2
    return 1


@app.get("/users/{user_id}")
async def get_user(user_id: str, version: int = Depends(get_api_version)):
    user = await user_repository.find_by_id(user_id)
    
    if version == 2:
        return {
            "id": user.id,
            "email": user.email,
            "first_name": user.first_name,
            "last_name": user.last_name,
        }
    else:
        return {
            "id": user.id,
            "email": user.email,
            "name": user.name,
        }

Versioning Best Practices

# Prefer additive changes (non-breaking)

# ✅ Non-breaking: Add new optional field
class UserV1(BaseModel):
    id: str
    email: str
    name: str


class UserV1Extended(BaseModel):
    id: str
    email: str
    name: str
    avatar_url: str | None = None  # New optional field


# ❌ Breaking: Change field type or remove field
# class UserV2(BaseModel):
#     id: int  # Changed from str to int - BREAKING!
#     email: str
#     # name removed - BREAKING!


# Deprecation strategy
from warnings import warn


@app.get("/users/{user_id}", deprecated=True)
async def get_user_deprecated(user_id: str):
    """
    Deprecated: Use /api/v2/users/{user_id} instead.
    This endpoint will be removed on 2025-06-01.
    """
    warn("Deprecated endpoint called", DeprecationWarning)
    return await get_user_v1(user_id)

Contract-First Design with OpenAPI

Defining the Contract First

# openapi/user-service.yaml
openapi: 3.1.0
info:
  title: User Service API
  version: 1.0.0
  description: User management microservice

paths:
  /users:
    get:
      operationId: listUsers
      summary: List all users
      parameters:
        - name: page
          in: query
          schema:
            type: integer
            default: 1
        - name: page_size
          in: query
          schema:
            type: integer
            default: 20
            maximum: 100
      responses:
        '200':
          description: List of users
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/UserListResponse'
    
    post:
      operationId: createUser
      summary: Create a new user
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/CreateUserRequest'
      responses:
        '201':
          description: User created
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/User'
        '422':
          description: Validation error
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/ErrorResponse'

  /users/{userId}:
    get:
      operationId: getUser
      summary: Get user by ID
      parameters:
        - name: userId
          in: path
          required: true
          schema:
            type: string
            format: uuid
      responses:
        '200':
          description: User found
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/User'
        '404':
          description: User not found
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/ErrorResponse'

components:
  schemas:
    User:
      type: object
      required:
        - id
        - email
        - name
        - created_at
      properties:
        id:
          type: string
          format: uuid
        email:
          type: string
          format: email
        name:
          type: string
          minLength: 1
          maxLength: 100
        created_at:
          type: string
          format: date-time
    
    CreateUserRequest:
      type: object
      required:
        - email
        - name
      properties:
        email:
          type: string
          format: email
        name:
          type: string
          minLength: 1
          maxLength: 100
    
    UserListResponse:
      type: object
      properties:
        data:
          type: array
          items:
            $ref: '#/components/schemas/User'
        meta:
          $ref: '#/components/schemas/PaginationMeta'
    
    PaginationMeta:
      type: object
      properties:
        page:
          type: integer
        page_size:
          type: integer
        total_items:
          type: integer
        total_pages:
          type: integer
    
    ErrorResponse:
      type: object
      properties:
        error:
          type: string
        message:
          type: string
        details:
          type: array
          items:
            $ref: '#/components/schemas/ErrorDetail'
        request_id:
          type: string
    
    ErrorDetail:
      type: object
      properties:
        field:
          type: string
        message:
          type: string
        code:
          type: string

Implementing from Contract

# Generate models from OpenAPI (optional: use datamodel-code-generator)
# pip install datamodel-code-generator
# datamodel-codegen --input openapi/user-service.yaml --output src/models/generated.py

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from uuid import UUID


# Models matching OpenAPI spec
class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str = Field(min_length=1, max_length=100)


class User(BaseModel):
    id: UUID
    email: EmailStr
    name: str
    created_at: datetime


class PaginationMeta(BaseModel):
    page: int
    page_size: int
    total_items: int
    total_pages: int


class UserListResponse(BaseModel):
    data: list[User]
    meta: PaginationMeta


# FastAPI app with OpenAPI customization
app = FastAPI(
    title="User Service API",
    version="1.0.0",
    description="User management microservice",
    openapi_url="/openapi.json",
    docs_url="/docs",
    redoc_url="/redoc",
)


@app.get("/users", response_model=UserListResponse, operation_id="listUsers")
async def list_users(page: int = 1, page_size: int = Field(default=20, le=100)):
    """List all users with pagination."""
    users, total = await user_repository.find_paginated(page, page_size)
    return UserListResponse(
        data=users,
        meta=PaginationMeta(
            page=page,
            page_size=page_size,
            total_items=total,
            total_pages=(total + page_size - 1) // page_size,
        ),
    )


@app.post(
    "/users",
    response_model=User,
    status_code=201,
    operation_id="createUser",
)
async def create_user(request: CreateUserRequest):
    """Create a new user."""
    user = await user_repository.create(
        email=request.email,
        name=request.name,
    )
    return user


@app.get("/users/{user_id}", response_model=User, operation_id="getUser")
async def get_user(user_id: UUID):
    """Get user by ID."""
    user = await user_repository.find_by_id(str(user_id))
    if not user:
        raise ResourceNotFoundError("User", str(user_id))
    return user

Consumer-Driven Contract Testing

# Contract tests ensure API compatibility
import pytest
from httpx import AsyncClient
from jsonschema import validate


# Define expected contract
USER_SCHEMA = {
    "type": "object",
    "required": ["id", "email", "name", "created_at"],
    "properties": {
        "id": {"type": "string", "format": "uuid"},
        "email": {"type": "string", "format": "email"},
        "name": {"type": "string"},
        "created_at": {"type": "string", "format": "date-time"},
    },
}


@pytest.mark.asyncio
async def test_get_user_contract():
    """Verify API response matches contract."""
    async with AsyncClient(base_url="http://localhost:8000") as client:
        # Create a user first
        create_response = await client.post(
            "/users",
            json={"email": "[email protected]", "name": "Test User"},
        )
        assert create_response.status_code == 201
        user_id = create_response.json()["id"]
        
        # Get the user
        response = await client.get(f"/users/{user_id}")
        assert response.status_code == 200
        
        # Validate against schema
        user_data = response.json()
        validate(instance=user_data, schema=USER_SCHEMA)


@pytest.mark.asyncio
async def test_list_users_contract():
    """Verify list endpoint matches contract."""
    async with AsyncClient(base_url="http://localhost:8000") as client:
        response = await client.get("/users")
        assert response.status_code == 200
        
        data = response.json()
        assert "data" in data
        assert "meta" in data
        assert isinstance(data["data"], list)
        
        if data["data"]:
            validate(instance=data["data"][0], schema=USER_SCHEMA)

API Documentation

from fastapi import FastAPI, Query, Path
from enum import Enum


class OrderStatus(str, Enum):
    pending = "pending"
    confirmed = "confirmed"
    shipped = "shipped"
    delivered = "delivered"


app = FastAPI(
    title="Order Service API",
    description="""
    ## Order Management Service
    
    This service handles order lifecycle management including:
    
    * Creating new orders
    * Tracking order status
    * Managing order items
    
    ### Authentication
    
    All endpoints require Bearer token authentication.
    
    ### Rate Limiting
    
    - 100 requests per minute for read operations
    - 10 requests per minute for write operations
    """,
    version="1.0.0",
    contact={
        "name": "API Support",
        "email": "[email protected]",
    },
    license_info={
        "name": "MIT",
    },
)


@app.get(
    "/orders/{order_id}",
    summary="Get order by ID",
    description="Retrieve detailed information about a specific order.",
    response_description="The order details",
    responses={
        200: {
            "description": "Order found",
            "content": {
                "application/json": {
                    "example": {
                        "id": "123e4567-e89b-12d3-a456-426614174000",
                        "status": "confirmed",
                        "total": 99.99,
                    }
                }
            },
        },
        404: {"description": "Order not found"},
    },
    tags=["Orders"],
)
async def get_order(
    order_id: str = Path(
        description="The unique identifier of the order",
        example="123e4567-e89b-12d3-a456-426614174000",
    ),
):
    """
    Get detailed information about an order.
    
    - **order_id**: UUID of the order to retrieve
    
    Returns the full order details including items and status.
    """
    pass


@app.get("/orders", tags=["Orders"])
async def list_orders(
    status: OrderStatus | None = Query(
        default=None,
        description="Filter orders by status",
    ),
    page: int = Query(
        default=1,
        ge=1,
        description="Page number for pagination",
    ),
    page_size: int = Query(
        default=20,
        ge=1,
        le=100,
        description="Number of items per page",
    ),
):
    """List orders with optional filtering and pagination."""
    pass

Practical Exercise

Exercise: Design an Order API

Design a complete API for an order service:

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from datetime import datetime
from decimal import Decimal
from uuid import UUID, uuid4
from enum import Enum


class OrderStatus(str, Enum):
    pending = "pending"
    confirmed = "confirmed"
    processing = "processing"
    shipped = "shipped"
    delivered = "delivered"
    cancelled = "cancelled"


class OrderItemCreate(BaseModel):
    product_id: UUID
    quantity: int = Field(ge=1)
    unit_price: Decimal = Field(ge=0)


class OrderItem(BaseModel):
    id: UUID
    product_id: UUID
    quantity: int
    unit_price: Decimal
    total: Decimal


class CreateOrderRequest(BaseModel):
    customer_id: UUID
    items: list[OrderItemCreate] = Field(min_length=1)
    shipping_address: str


class Order(BaseModel):
    id: UUID
    customer_id: UUID
    status: OrderStatus
    items: list[OrderItem]
    subtotal: Decimal
    tax: Decimal
    total: Decimal
    shipping_address: str
    created_at: datetime
    updated_at: datetime


class UpdateOrderStatusRequest(BaseModel):
    status: OrderStatus


app = FastAPI(title="Order Service")


@app.post("/orders", response_model=Order, status_code=201)
async def create_order(request: CreateOrderRequest):
    """Create a new order."""
    order_id = uuid4()
    items = []
    subtotal = Decimal("0")
    
    for item_data in request.items:
        item_total = item_data.unit_price * item_data.quantity
        items.append(OrderItem(
            id=uuid4(),
            product_id=item_data.product_id,
            quantity=item_data.quantity,
            unit_price=item_data.unit_price,
            total=item_total,
        ))
        subtotal += item_total
    
    tax = subtotal * Decimal("0.1")  # 10% tax
    
    return Order(
        id=order_id,
        customer_id=request.customer_id,
        status=OrderStatus.pending,
        items=items,
        subtotal=subtotal,
        tax=tax,
        total=subtotal + tax,
        shipping_address=request.shipping_address,
        created_at=datetime.utcnow(),
        updated_at=datetime.utcnow(),
    )


@app.get("/orders/{order_id}", response_model=Order)
async def get_order(order_id: UUID):
    """Get order by ID."""
    # Implementation here
    pass


@app.patch("/orders/{order_id}/status", response_model=Order)
async def update_order_status(order_id: UUID, request: UpdateOrderStatusRequest):
    """Update order status."""
    # Implementation here
    pass


@app.post("/orders/{order_id}/cancel", response_model=Order)
async def cancel_order(order_id: UUID):
    """Cancel an order. Only pending orders can be cancelled."""
    # Implementation here
    pass

Key Takeaways

1. Use resources, not actions - Design around nouns, use HTTP methods for verbs

2. Be consistent - Same patterns across all endpoints

3. Version from day one - Plan for breaking changes

4. Contract-first - Define OpenAPI spec before implementation

5. Document thoroughly - Good docs reduce support burden

What's Next?

With well-designed APIs, we need to understand how services communicate. In Article 4: Synchronous Communication, we'll cover HTTP/REST patterns, gRPC, and service discovery.

---

This article is part of the Microservice Architecture 101 series.