---
name: api-design
description: REST API design patterns with Pydantic/Zod schemas, error handling, and OpenAPI documentation. Use when designing or implementing API endpoints.
---

# API Design Skill

## Schema-First Development

Always define schemas before implementation. Schemas serve as:
- Runtime validation
- Type definitions
- API documentation
- Test data factories

### Python (Pydantic)

```python
# schemas/user.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field, field_validator


class UserBase(BaseModel):
    """Shared fields for user schemas."""
    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)


class UserCreate(UserBase):
    """Request schema for creating a user."""
    password: str = Field(..., min_length=8)

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain uppercase")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password must contain digit")
        return v


class UserUpdate(BaseModel):
    """Request schema for updating a user (all optional)."""
    email: EmailStr | None = None
    name: str | None = Field(None, min_length=1, max_length=100)


class UserResponse(UserBase):
    """Response schema (no password)."""
    id: str
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}


class UserListResponse(BaseModel):
    """Paginated list response."""
    items: list[UserResponse]
    total: int
    page: int
    page_size: int
    has_more: bool
```

### TypeScript (Zod)

```typescript
// schemas/user.schema.ts
import { z } from 'zod';

export const userBaseSchema = z.object({
  email: z.string().email(),
  name: z.string().min(1).max(100),
});

export const userCreateSchema = userBaseSchema.extend({
  password: z
    .string()
    .min(8)
    .refine((p) => /[A-Z]/.test(p), 'Must contain uppercase')
    .refine((p) => /\d/.test(p), 'Must contain digit'),
});

export const userUpdateSchema = userBaseSchema.partial();

export const userResponseSchema = userBaseSchema.extend({
  id: z.string().uuid(),
  isActive: z.boolean(),
  createdAt: z.string().datetime(),
});

export const userListResponseSchema = z.object({
  items: z.array(userResponseSchema),
  total: z.number().int().nonnegative(),
  page: z.number().int().positive(),
  pageSize: z.number().int().positive(),
  hasMore: z.boolean(),
});

// Derived types
export type UserCreate = z.infer<typeof userCreateSchema>;
export type UserUpdate = z.infer<typeof userUpdateSchema>;
export type UserResponse = z.infer<typeof userResponseSchema>;
export type UserListResponse = z.infer<typeof userListResponseSchema>;

// Validation functions for API boundaries
export const parseUserCreate = (data: unknown) => userCreateSchema.parse(data);
export const parseUserResponse = (data: unknown) => userResponseSchema.parse(data);
```

## REST Endpoint Patterns

### Resource Naming
```
GET    /users              # List users
POST   /users              # Create user
GET    /users/{id}         # Get single user
PUT    /users/{id}         # Full update
PATCH  /users/{id}         # Partial update
DELETE /users/{id}         # Delete user

# Nested resources
GET    /users/{id}/orders  # User's orders
POST   /users/{id}/orders  # Create order for user

# Actions (when CRUD doesn't fit)
POST   /users/{id}/activate
POST   /orders/{id}/cancel
```

### FastAPI Implementation

```python
# routers/users.py
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.schemas.user import (
    UserCreate,
    UserUpdate,
    UserResponse,
    UserListResponse,
)
from app.services.user import UserService
from app.dependencies import get_db, get_current_user

router = APIRouter(prefix="/users", tags=["users"])


@router.get("", response_model=UserListResponse)
async def list_users(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
) -> UserListResponse:
    """List users with pagination."""
    service = UserService(db)
    return await service.list_users(page=page, page_size=page_size)


@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    data: UserCreate,
    db: AsyncSession = Depends(get_db),
) -> UserResponse:
    """Create a new user."""
    service = UserService(db)
    try:
        return await service.create_user(data)
    except ValueError as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: str,
    db: AsyncSession = Depends(get_db),
) -> UserResponse:
    """Get a user by ID."""
    service = UserService(db)
    user = await service.get_user(user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user


@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: str,
    data: UserUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: UserResponse = Depends(get_current_user),
) -> UserResponse:
    """Partially update a user."""
    service = UserService(db)
    user = await service.update_user(user_id, data)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user


@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    user_id: str,
    db: AsyncSession = Depends(get_db),
    current_user: UserResponse = Depends(get_current_user),
) -> None:
    """Delete a user."""
    service = UserService(db)
    deleted = await service.delete_user(user_id)
    if not deleted:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
```

## Error Handling

### Standard Error Response (RFC 7807)

```python
# schemas/error.py
from pydantic import BaseModel


class ErrorDetail(BaseModel):
    """Standard error response following RFC 7807."""
    type: str = "about:blank"
    title: str
    status: int
    detail: str
    instance: str | None = None


# Exception handler
from fastapi import Request
from fastapi.responses import JSONResponse

async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content=ErrorDetail(
            type="validation_error",
            title="Validation Error",
            status=422,
            detail=str(exc.errors()),
            instance=str(request.url),
        ).model_dump(),
    )
```

### TypeScript Error Handling

```typescript
// lib/api-client.ts
import axios, { AxiosError } from 'axios';
import { z } from 'zod';

const errorSchema = z.object({
  type: z.string(),
  title: z.string(),
  status: z.number(),
  detail: z.string(),
  instance: z.string().optional(),
});

export class ApiError extends Error {
  constructor(
    public status: number,
    public title: string,
    public detail: string,
  ) {
    super(detail);
    this.name = 'ApiError';
  }
}

export const apiClient = axios.create({
  baseURL: '/api',
  headers: { 'Content-Type': 'application/json' },
});

apiClient.interceptors.response.use(
  (response) => response,
  (error: AxiosError) => {
    if (error.response?.data) {
      const parsed = errorSchema.safeParse(error.response.data);
      if (parsed.success) {
        throw new ApiError(
          parsed.data.status,
          parsed.data.title,
          parsed.data.detail,
        );
      }
    }
    throw new ApiError(500, 'Server Error', 'An unexpected error occurred');
  },
);
```

## Pagination Pattern

```python
# schemas/pagination.py
from typing import Generic, TypeVar
from pydantic import BaseModel, Field

T = TypeVar("T")


class PaginatedResponse(BaseModel, Generic[T]):
    """Generic paginated response."""
    items: list[T]
    total: int
    page: int = Field(ge=1)
    page_size: int = Field(ge=1, le=100)

    @property
    def has_more(self) -> bool:
        return self.page * self.page_size < self.total

    @property
    def total_pages(self) -> int:
        return (self.total + self.page_size - 1) // self.page_size


# Usage
class UserListResponse(PaginatedResponse[UserResponse]):
    pass
```

## Query Parameters

```python
# dependencies/pagination.py
from fastapi import Query
from pydantic import BaseModel


class PaginationParams(BaseModel):
    page: int = Query(1, ge=1, description="Page number")
    page_size: int = Query(20, ge=1, le=100, description="Items per page")

    @property
    def offset(self) -> int:
        return (self.page - 1) * self.page_size


class SortParams(BaseModel):
    sort_by: str = Query("created_at", description="Field to sort by")
    sort_order: str = Query("desc", pattern="^(asc|desc)$")


class FilterParams(BaseModel):
    search: str | None = Query(None, min_length=1, max_length=100)
    status: str | None = Query(None, pattern="^(active|inactive|pending)$")
    created_after: datetime | None = Query(None)
    created_before: datetime | None = Query(None)
```

## OpenAPI Documentation

```python
# main.py
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI(
    title="My API",
    description="API for managing resources",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title=app.title,
        version=app.version,
        description=app.description,
        routes=app.routes,
    )

    # Add security scheme
    openapi_schema["components"]["securitySchemes"] = {
        "bearerAuth": {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "JWT",
        }
    }

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi
```

## HTTP Status Codes

| Code | Meaning | When to Use |
|------|---------|-------------|
| 200 | OK | Successful GET, PUT, PATCH |
| 201 | Created | Successful POST creating resource |
| 204 | No Content | Successful DELETE |
| 400 | Bad Request | Invalid request body/params |
| 401 | Unauthorized | Missing/invalid authentication |
| 403 | Forbidden | Authenticated but not authorized |
| 404 | Not Found | Resource doesn't exist |
| 409 | Conflict | Duplicate resource (e.g., email exists) |
| 422 | Unprocessable | Validation error |
| 500 | Server Error | Unexpected server error |

## Anti-Patterns

```python
# BAD: Returning different shapes
@router.get("/users/{id}")
async def get_user(id: str):
    user = await get_user(id)
    if user:
        return user  # UserResponse
    return {"error": "not found"}  # Different shape!

# GOOD: Consistent response or exception
@router.get("/users/{id}", response_model=UserResponse)
async def get_user(id: str):
    user = await get_user(id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


# BAD: Exposing internal details
class UserResponse(BaseModel):
    id: str
    email: str
    hashed_password: str  # NEVER expose!
    internal_notes: str   # Internal only!

# GOOD: Explicit public fields
class UserResponse(BaseModel):
    id: str
    email: str
    name: str
    # Only fields clients need


# BAD: No validation at boundary
@router.post("/users")
async def create_user(data: dict):  # Unvalidated!
    return await service.create(data)

# GOOD: Schema validation
@router.post("/users", response_model=UserResponse)
async def create_user(data: UserCreate):  # Validated!
    return await service.create(data)
```