Dev Api Design

API Development & Design — Quick Reference

Use this skill to design, implement, and document production-grade APIs (REST, GraphQL, gRPC, and tRPC). Apply it for contract design (OpenAPI), versioning/deprecation, authentication/authorization, rate limiting, pagination, error models, and developer documentation.

Modern best practices (Jan 2026): HTTP semantics and cacheability (RFC 9110), Problem Details error model (RFC 9457), OpenAPI 3.1+, contract-first + breaking-change detection, strong AuthN/Z boundaries, explicit versioning/deprecation, and operable-by-default APIs (idempotency, rate limits, observability, trace context).

---

Default Execution Checklist

• Choose an API style based on constraints (public vs internal, performance, client query flexibility).
• Define the contract first (OpenAPI or GraphQL schema; protobuf for gRPC).
• Define the error model (RFC 9457 + stable error codes + trace IDs).
• Define AuthN/AuthZ boundaries (scopes/roles/tenancy) and threat model.
• Define pagination/filter/sort for all list endpoints.
• Define rate limits/quotas, idempotency strategy (esp. POST), and retries/backoff guidance.
• Define observability (W3C Trace Context, request IDs, metrics, logs) and SLOs.
• Add contract tests + breaking-change checks in CI.
• Publish docs with examples + migration/deprecation policy.

---

Quick Reference

---

Decision Tree: Choosing API Style

User needs: [API Type]
    ├─ Public API for third parties?
    │   └─ REST with OpenAPI docs (broad compatibility)
    │
    ├─ Internal microservices?
    │   ├─ High throughput required? → **gRPC** (binary, fast)
    │   └─ Simple CRUD? → **REST** (easy to debug)
    │
    ├─ TypeScript monorepo (frontend + backend)?
    │   └─ **tRPC** (end-to-end type safety, no codegen)
    │
    ├─ Client needs flexible queries?
    │   ├─ Real-time updates? → **GraphQL Subscriptions** or **WebSockets**
    │   └─ Complex data fetching? → **GraphQL** (avoid over-fetching)
    │
    ├─ Mobile/web clients?
    │   ├─ Many entity types? → **GraphQL** (single endpoint)
    │   └─ Simple resources? → **REST** (cacheable)
    │
    ├─ AI agents consuming API?
    │   └─ REST + **MCP** wrapper (agent experience)
    │
    └─ Streaming or bidirectional?
        └─ **gRPC** (HTTP/2 streaming) or **WebSockets**

---

Navigation: Core API Patterns

RESTful API Design

Resource: references/restful-design-patterns.md

• Resource-based URLs with proper HTTP methods (GET, POST, PUT, PATCH, DELETE)
• HTTP status code semantics (200, 201, 404, 422, 500)
• Idempotency guarantees (GET, PUT, DELETE)
• Stateless design principles
• URL structure best practices (collection vs resource endpoints)
• Nested resources and action endpoints

---

Pagination, Filtering & Sorting

Resource: references/pagination-filtering.md

• Offset-based pagination (simple, static datasets)
• Cursor-based pagination (real-time feeds, recommended)
• Page-based pagination (UI with page numbers)
• Query parameter filtering with operators (_gt, _contains, _in)
• Multi-field sorting with direction (-created_at)
• Performance optimization with indexes

---

Error Handling

Resource: references/error-handling-patterns.md

• RFC 9457 Problem Details standard
• HTTP status code reference (4xx client errors, 5xx server errors)
• Field-level validation errors
• Trace IDs for debugging
• Consistent error format across endpoints
• Security-safe error messages (no stack traces in production)

---

Authentication & Authorization

Resource: references/authentication-patterns.md

• JWT (JSON Web Tokens) with refresh token rotation
• OAuth2 Authorization Code Flow for third-party auth
• API Key authentication for server-to-server
• RBAC (Role-Based Access Control)
• ABAC (Attribute-Based Access Control)
• Resource-based authorization (user-owned resources)

---

Rate Limiting & Throttling

Resource: references/rate-limiting-patterns.md

• Token Bucket algorithm (recommended, allows bursts)
• Fixed Window vs Sliding Window
• Rate limit headers (X-RateLimit-*)
• Tiered rate limits (free, paid, enterprise)
• Redis-based distributed rate limiting
• Per-user, per-endpoint, and per-API-key strategies

---

Navigation: Extended Resources

API Design & Best Practices

• [api-design-best-practices.md](references/api-design-best-practices.md) - Comprehensive API design principles
• [versioning-strategies.md](references/versioning-strategies.md) - URL, header, and query parameter versioning
• [api-security-checklist.md](references/api-security-checklist.md) - OWASP API Security Top 10

GraphQL & gRPC

• [graphql-patterns.md](references/graphql-patterns.md) - Schema design, resolvers, N+1 queries, DataLoader
• gRPC patterns - See software-backend for Protocol Buffers and service definitions

tRPC (TypeScript-First)

• [trpc-patterns.md](references/trpc-patterns.md) - End-to-end type safety, procedures, React Query integration
• When to use tRPC vs GraphQL vs REST
• Auth middleware patterns
• Server-side rendering with Next.js

OpenAPI & Documentation

• [openapi-guide.md](references/openapi-guide.md) - OpenAPI 3.1 specifications, Swagger UI, Redoc
• Templates: assets/openapi-template.yaml - Complete OpenAPI spec example

Webhooks & Event-Driven APIs

• [webhook-patterns.md](references/webhook-patterns.md) - Webhook design, delivery guarantees, signature verification, retry policies, DLQs

Real-Time APIs

• [real-time-api-patterns.md](references/real-time-api-patterns.md) - WebSockets, SSE, long polling, gRPC streaming, protocol selection guide

API Testing

• [api-testing-patterns.md](references/api-testing-patterns.md) - Contract testing, integration testing, load testing, chaos testing for APIs

Optional: AI/Automation (LLM/Agent APIs)

• [llm-agent-api-contracts.md](references/llm-agent-api-contracts.md) - Streaming, long-running jobs, safety guardrails, observability

---

Navigation: Templates

Production-ready, copy-paste API implementations with authentication, database, validation, and docs.

Framework-Specific Templates

• FastAPI (Python): assets/fastapi/fastapi-complete-api.md
• Async/await, Pydantic v2, JWT auth, SQLAlchemy 2.0, pagination, OpenAPI docs

• Express.js (Node/TypeScript): assets/express-nodejs/express-complete-api.md
• TypeScript, Zod validation, Prisma ORM, JWT refresh tokens, rate limiting

• Django REST Framework: assets/django-rest/django-rest-complete-api.md
• ViewSets, serializers, Simple JWT, permissions, DRF filtering/pagination

• Spring Boot (Java): assets/spring-boot/spring-boot-complete-api.md
• Spring Security JWT, Spring Data JPA, Bean Validation, Springdoc OpenAPI

Cross-Platform Patterns

• [api-patterns-universal.md](assets/cross-platform/api-patterns-universal.md) - Universal patterns for all frameworks
• Authentication strategies, pagination, caching, versioning, validation
• [template-api-governance.md](assets/cross-platform/template-api-governance.md) - API governance, deprecation, multi-tenancy
• Deprecation policy (90-day timeline), backward compatibility rules, error model templates
• [template-api-design-review-checklist.md](assets/cross-platform/template-api-design-review-checklist.md) - Production API review checklist (security, reliability, operability)
• [template-api-error-model.md](assets/cross-platform/template-api-error-model.md) - RFC 9457 Problem Details + stable error code registry

---

Do / Avoid

GOOD: Do

• Version APIs from day one
• Document deprecation policy before first deprecation
• Treat breaking changes as a major version (and keep minor changes backward compatible)
• Include trace IDs in all error responses
• Return appropriate HTTP status codes
• Implement rate limiting with clear headers
• Use RFC 9457 Problem Details for errors

BAD: Avoid

• Removing fields without deprecation period
• Changing field types in existing versions
• Using verbs in resource names (nouns only)
• Returning 500 for client errors
• Breaking changes without major version bump
• Mixing tenant data without explicit isolation
• Action endpoints everywhere (/doSomething)

---

Anti-Patterns

---

Optional: AI/Automation

Note: AI tools assist but contracts need human review.

• OpenAPI linting — Spectral, Redocly in CI/CD
• Breaking change detection — oasdiff automated checks
• SDK generation — From OpenAPI spec on changes
• Contract testing — Pact, Dredd automation

Bounded Claims

• AI-generated OpenAPI specs require human review
• Automated deprecation detection needs manual confirmation
• SDK generation requires type verification

---

External Resources

See data/sources.json for:

• Official REST, GraphQL, gRPC documentation
• OpenAPI/Swagger tools and validators
• API design style guides (Google, Microsoft, Stripe)
• Security standards (OWASP API Security Top 10)
• Testing tools (Postman, Insomnia, Paw)

---

Related Skills

This skill works best when combined with other specialized skills:

Backend Development

• [software-backend](../software-backend/SKILL.md) - Production backend patterns (Node.js, Python, Java frameworks)
• Use when implementing API server infrastructure
• Covers database integration, middleware, error handling

Security & Authentication

• [software-security-appsec](../software-security-appsec/SKILL.md) - Application security patterns
• Critical for securing API endpoints
• Covers OWASP vulnerabilities, authentication flows, input validation

Database & Data Layer

• [data-sql-optimization](../data-sql-optimization/SKILL.md) - SQL optimization and database patterns
• Essential for API performance (query optimization, indexing)
• Use when APIs interact with relational databases

Testing & Quality

• [qa-testing-strategy](../qa-testing-strategy/SKILL.md) - Test strategy and automation
• Contract testing for API specifications
• Integration testing for API endpoints

DevOps & Deployment

• [ops-devops-platform](../ops-devops-platform/SKILL.md) - Platform engineering and deployment
• API gateway configuration
• CI/CD pipelines for API deployments

Documentation

• [docs-codebase](../docs-codebase/SKILL.md) - Technical documentation standards
• API reference documentation structure
• Complements OpenAPI auto-generated docs

Architecture

• [software-architecture-design](../software-architecture-design/SKILL.md) - System design patterns
• Microservices architecture with APIs
• API gateway patterns, service mesh integration

Performance & Observability

• [qa-observability](../qa-observability/SKILL.md) - Performance optimization and monitoring
• API latency monitoring, distributed tracing
• Performance budgets for API endpoints

---

Usage Notes

For the agent:

• Apply RESTful principles by default unless user requests GraphQL/gRPC
• Always include pagination for list endpoints
• Use RFC 9457 format for error responses
• Include authentication in all templates (JWT or API keys)
• Reference framework-specific templates for complete implementations
• Link to relevant resources for deep-dive guidance

Success Criteria: APIs are discoverable, consistent, well-documented, secure, and follow HTTP/GraphQL semantics correctly.

---

Time-Sensitive Recommendations

If a user asks for "best" tools/frameworks, "latest" standards, or whether something is still relevant in 2026, do a quick web search using whatever browsing/search tool is available in the current environment. If web access is unavailable, answer from stable principles, state assumptions (traffic, latency, team skills, ecosystem), and avoid overstating currency.

Fact-Checking

• Use web search/web fetch to verify current external facts, versions, pricing, deadlines, regulations, or platform behavior before final answers.
• Prefer primary sources; report source links and dates for volatile information.
• If web access is unavailable, state the limitation and mark guidance as unverified.

Universal API Design Patterns

Cross-framework patterns and best practices applicable to REST, GraphQL, and gRPC APIs regardless of implementation stack.

Table of Contents

1. Authentication Patterns 2. Authorization Patterns 3. Error Handling 4. Pagination Strategies 5. Rate Limiting 6. Caching Strategies 7. API Versioning 8. Input Validation 9. Response Formatting 10. Performance Optimization

---

Authentication Patterns

Pattern 1: JWT (JSON Web Tokens)

Use when: Stateless authentication needed, distributed systems, microservices

Structure:

Header.Payload.Signature

Payload Example:

{
  "sub": "user-id",
  "email": "user@example.com",
  "role": "admin",
  "iat": 1640000000,
  "exp": 1640003600
}

Best Practices:

• Short access token lifetime (15-60 minutes)
• Long refresh token lifetime (7-30 days)
• Store refresh tokens securely (httpOnly cookies or secure storage)
• Rotate refresh tokens on each use
• Include minimal claims in payload
• Use strong signing algorithms (RS256, HS256)

Security Checklist:

• [ ] Token expiration enforced
• [ ] Signature verification on every request
• [ ] Sensitive data not in payload
• [ ] HTTPS only for token transmission
• [ ] Token revocation strategy implemented
• [ ] Rate limiting on token endpoints

Pattern 2: OAuth 2.0 Flows

Authorization Code Flow (Recommended for Web Apps):

1. Client → Authorization Server: GET /authorize?client_id=...&redirect_uri=...&scope=...
2. User authenticates and grants permission
3. Authorization Server → Client: Redirect with authorization code
4. Client → Authorization Server: POST /token with code + client_secret
5. Authorization Server → Client: Access token + refresh token
6. Client → Resource Server: Request with Bearer token

Client Credentials Flow (Service-to-Service):

1. Service → Authorization Server: POST /token with client_id + client_secret
2. Authorization Server → Service: Access token
3. Service → API: Request with Bearer token

Best Practices:

• Use PKCE (Proof Key for Code Exchange) for public clients
• Implement state parameter for CSRF protection
• Store client secrets securely (never in frontend)
• Scope tokens to minimum required permissions
• Use short-lived access tokens with refresh rotation

Pattern 3: API Key Authentication

Use when: Simple service-to-service communication, webhooks, public APIs with usage tracking

Implementation Options:

Option 1: Header-based

GET /api/v1/resources
X-API-Key: your-api-key-here

Option 2: Query parameter (not recommended for sensitive data)

GET /api/v1/resources?api_key=your-api-key-here

Best Practices:

• Generate cryptographically random keys (32+ characters)
• Support multiple keys per user/service
• Enable key rotation without downtime
• Track usage per API key
• Allow key expiration and revocation
• Never log API keys
• Store hashed versions only

Rate Limiting Per Key:

Free tier:     100 requests/hour
Paid tier:    1000 requests/hour
Enterprise:  10000 requests/hour

---

Authorization Patterns

Pattern 1: Role-Based Access Control (RBAC)

Use when: Fixed set of roles, hierarchical permissions

Structure:

User → Role → Permissions

Example:
- Admin: [read, write, delete, manage_users]
- Editor: [read, write]
- Viewer: [read]

Implementation:

# Pseudocode
def check_permission(user, action, resource):
    user_roles = get_user_roles(user)
    required_permissions = get_required_permissions(action, resource)

    for role in user_roles:
        if role.has_permission(required_permissions):
            return True
    return False

Best Practices:

• Check permissions on every request
• Deny by default
• Least privilege principle
• Audit permission changes
• Cache role lookups

Pattern 2: Attribute-Based Access Control (ABAC)

Use when: Complex, context-dependent permissions

Policy Example:

{
  "policy": "Allow user to edit document if they are the owner OR they are in the same department AND document is not locked",
  "conditions": {
    "owner": "user.id == document.owner_id",
    "department": "user.department == document.department",
    "not_locked": "document.status != 'locked'"
  }
}

Best Practices:

• Define policies as code
• Cache policy evaluations
• Log policy decisions
• Test policies thoroughly

Pattern 3: Resource-Based Authorization

Use when: Users own resources, multi-tenancy

Check:

if (resource.owner_id !== current_user.id && !current_user.is_admin) {
    throw Forbidden("Not authorized to access this resource");
}

Multi-Tenancy Pattern:

SELECT * FROM resources
WHERE tenant_id = :current_tenant_id
  AND (owner_id = :current_user_id OR is_public = true);

---

Error Handling

RFC 9457 Problem Details Format (Obsoletes RFC 7807)

Standard Error Response:

{
  "type": "https://api.example.com/errors/validation-error",
  "title": "Validation Error",
  "status": 422,
  "detail": "Email address is already registered",
  "instance": "/api/v1/users",
  "code": "DUPLICATE_EMAIL",
  "retryable": false,
  "errors": [
    {
      "field": "email",
      "code": "DUPLICATE_EMAIL",
      "message": "Email address is already registered"
    }
  ],
  "traceId": "abc123def456"
}

HTTP Status Code Guide

Success (2xx):

• 200 OK - Successful GET, PUT, PATCH
• 201 Created - Successful POST (include Location header)
• 204 No Content - Successful DELETE or operation with no return

Client Errors (4xx):

• 400 Bad Request - Malformed request syntax
• 401 Unauthorized - Authentication required or failed
• 403 Forbidden - Authenticated but insufficient permissions
• 404 Not Found - Resource not found
• 409 Conflict - Duplicate resource or state conflict
• 422 Unprocessable Entity - Validation error
• 429 Too Many Requests - Rate limit exceeded

Server Errors (5xx):

• 500 Internal Server Error - Unexpected server error
• 502 Bad Gateway - Upstream service error
• 503 Service Unavailable - Temporary downtime
• 504 Gateway Timeout - Upstream timeout

Error Handling Best Practices

Production vs Development:

if (environment === 'production') {
    return {
        status: 500,
        detail: "An unexpected error occurred"
    };
} else {
    return {
        status: 500,
        detail: error.message,
        stack: error.stack
    };
}

Checklist:

• [ ] Consistent error format across all endpoints
• [ ] Machine-readable error codes
• [ ] Human-readable messages
• [ ] Field-level validation errors
• [ ] Trace IDs for debugging
• [ ] No sensitive data in responses
• [ ] Different messages for dev vs prod

---

Pagination Strategies

Strategy 1: Offset-Based Pagination

Use when: Displaying page numbers, static data

Request:

GET /api/v1/users?limit=20&offset=40

Response:

{
  "data": [...],
  "meta": {
    "total": 1500,
    "limit": 20,
    "offset": 40,
    "hasMore": true,
    "totalPages": 75,
    "currentPage": 3
  }
}

Pros:

• Simple to implement
• Jump to any page
• Total count available

Cons:

• Performance degrades with large offsets
• Inconsistent results if data changes
• Not suitable for real-time feeds

Strategy 2: Cursor-Based Pagination

Use when: Real-time data, infinite scroll, large datasets

Request:

GET /api/v1/users?limit=20&cursor=eyJpZCI6MTIzLCJjcmVhdGVkX2F0IjoiMjAyNS0wMS0xNVQxMDowMDowMFoifQ==

Response:

{
  "data": [...],
  "meta": {
    "nextCursor": "eyJpZCI6MTQzLCJjcmVhdGVkX2F0IjoiMjAyNS0wMS0xNVQxMTowMDowMFoifQ==",
    "hasMore": true
  }
}

Cursor Structure (Base64 encoded):

{
  "id": 143,
  "created_at": "2025-01-15T11:00:00Z"
}

SQL Implementation:

SELECT * FROM users
WHERE (created_at, id) < (:cursor_created_at, :cursor_id)
ORDER BY created_at DESC, id DESC
LIMIT :limit;

Pros:

• Consistent results
• Excellent performance
• Real-time data support

Cons:

• Cannot jump to specific page
• No total count
• More complex implementation

Strategy 3: Keyset Pagination

Use when: Efficient pagination on indexed columns

Request:

GET /api/v1/users?limit=20&after_id=100

SQL:

SELECT * FROM users
WHERE id > :after_id
ORDER BY id
LIMIT :limit;

Best Practices:

• Default limit (e.g., 20)
• Maximum limit (e.g., 100)
• Include pagination metadata
• Document pagination strategy
• Validate cursor integrity
• Use indexed fields for cursors

---

Rate Limiting

Implementation Strategies

1. Token Bucket Algorithm (Recommended)

Concept:

• Bucket holds tokens
• Each request consumes a token
• Tokens refill at fixed rate
• Allows burst traffic

Parameters:

capacity: 100 tokens
refill_rate: 100 tokens/minute

Headers:

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 73
X-RateLimit-Reset: 1640000060
Retry-After: 60

429 Response:

{
  "type": "https://api.example.com/errors/rate-limit",
  "title": "Rate Limit Exceeded",
  "status": 429,
  "detail": "Too many requests",
  "retryAfter": 60
}

2. Fixed Window Counter

Simple but less accurate:

Window: 1 minute (00:00-00:59)
Max requests: 100
Reset: At minute boundary

3. Sliding Window Log

Most accurate but memory-intensive:

Track timestamp of each request
Count requests in last N seconds
Remove expired timestamps

Tiered Rate Limits

Free tier:     100 requests/hour   (per API key)
Paid tier:    1000 requests/hour   (per API key)
Enterprise:  10000 requests/hour   (per API key)
Per IP:        60 requests/minute  (global)

Best Practices

• [ ] Rate limit headers in all responses
• [ ] Different limits for different endpoints
• [ ] Per-user and per-IP limits
• [ ] Burst allowance for spiky traffic
• [ ] Graceful degradation
• [ ] Monitor and alert on limit hits
• [ ] Document limits clearly

---

Caching Strategies

HTTP Caching Headers

Cache-Control:

Cache-Control: public, max-age=3600
Cache-Control: private, max-age=600
Cache-Control: no-cache
Cache-Control: no-store

ETag (Entity Tag):

# Response
ETag: "33a64df551425fcc55e4d42a148795d9f25f89d4"
Cache-Control: max-age=3600

# Subsequent request
If-None-Match: "33a64df551425fcc55e4d42a148795d9f25f89d4"

# Response (if not modified)
HTTP/1.1 304 Not Modified

Last-Modified:

# Response
Last-Modified: Wed, 15 Jan 2025 10:00:00 GMT

# Subsequent request
If-Modified-Since: Wed, 15 Jan 2025 10:00:00 GMT

# Response (if not modified)
HTTP/1.1 304 Not Modified

Server-Side Caching

1. Application Cache (Redis/Memcached)

Pattern:

cache_key = f"user:{user_id}"
cached_data = cache.get(cache_key)

if cached_data:
    return cached_data

data = database.query(user_id)
cache.set(cache_key, data, ttl=600)  # 10 minutes
return data

2. Cache Invalidation Strategies

Time-based (TTL):

cache.set(key, value, ttl=3600)

Event-based:

def update_user(user_id, data):
    database.update(user_id, data)
    cache.delete(f"user:{user_id}")

Cache-Aside Pattern:

1. Check cache
2. If miss, query database
3. Store in cache
4. Return data

Best Practices

• [ ] Cache immutable data aggressively
• [ ] Short TTL for frequently changing data
• [ ] Include cache headers in responses
• [ ] Invalidate cache on writes
• [ ] Monitor cache hit rates
• [ ] Use versioned cache keys
• [ ] Handle cache failures gracefully

---

API Versioning

Strategy 1: URL Versioning (Recommended)

Example:

/api/v1/users
/api/v2/users

Pros:

• Explicit and visible
• Easy to route
• Cache-friendly
• Browser-testable

Cons:

• URL proliferation
• Clients must update URLs

Strategy 2: Header Versioning

Example:

GET /api/users
Accept: application/vnd.api+json; version=2

Pros:

• Clean URLs
• Content negotiation

Cons:

• Less visible
• Harder to test

Strategy 3: Query Parameter

Example:

/api/users?version=2

Pros:

• Simple
• Backward compatible

Cons:

• Pollutes query space
• Not RESTful

Versioning Best Practices

• [ ] Version only on breaking changes
• [ ] Support N-1 versions (current + previous)
• [ ] Deprecation policy (6-12 months notice)
• [ ] Sunset headers for deprecated versions
• [ ] Migration guides documented
• [ ] Changelog maintained

Sunset Header:

Sunset: Sat, 01 Jan 2026 00:00:00 GMT
Link: <https://api.example.com/docs/migration>; rel="deprecation"

---

Input Validation

Validation Layers

1. Schema Validation (Structure)

{
  "email": "string, required, format: email",
  "age": "number, optional, min: 0, max: 150"
}

2. Business Logic Validation

- Email uniqueness
- Password strength
- Date range validity
- Referential integrity

Validation Best Practices

Whitelist vs Blacklist:

// Good: Whitelist
const allowedFields = ['name', 'email'];
const data = pick(request.body, allowedFields);

// Bad: Blacklist
const data = omit(request.body, ['admin', 'password_hash']);

Sanitization:

const sanitized = {
    email: input.email.toLowerCase().trim(),
    name: stripHtml(input.name),
    age: parseInt(input.age)
};

Error Messages:

{
  "errors": [
    {
      "field": "email",
      "code": "INVALID_FORMAT",
      "message": "Email must be a valid email address"
    },
    {
      "field": "password",
      "code": "TOO_SHORT",
      "message": "Password must be at least 8 characters"
    }
  ]
}

Checklist

• [ ] Validate all inputs
• [ ] Fail fast (validate early)
• [ ] Specific error messages
• [ ] Sanitize before storage
• [ ] Reject unknown fields
• [ ] Type coercion documented
• [ ] Max request size enforced

---

Response Formatting

Consistent Response Structure

Success Response:

{
  "data": {
    "id": "123",
    "name": "John Doe"
  }
}

List Response:

{
  "data": [...],
  "meta": {
    "total": 1500,
    "limit": 20,
    "offset": 40
  }
}

Error Response:

{
  "type": "https://api.example.com/errors/validation",
  "title": "Validation Error",
  "status": 422,
  "detail": "Request validation failed",
  "errors": [...]
}

Field Naming Conventions

camelCase (JavaScript/TypeScript):

{
  "userId": "123",
  "createdAt": "2025-01-15T10:00:00Z"
}

snake_case (Python/Ruby):

{
  "user_id": "123",
  "created_at": "2025-01-15T10:00:00Z"
}

Consistency is key - choose one and stick with it.

Date/Time Formatting

Use ISO 8601:

{
  "created_at": "2025-01-15T10:30:00Z",
  "updated_at": "2025-01-15T14:30:00+02:00"
}

---

Performance Optimization

1. Database Query Optimization

N+1 Query Problem:

// Bad: N+1 queries
const users = await User.findAll();
for (const user of users) {
    user.orders = await Order.findByUserId(user.id);  // N queries
}

// Good: Join or eager loading
const users = await User.findAll({
    include: [{ model: Order }]  // 1 query
});

Indexing:

CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_orders_user_id ON orders(user_id);

2. Response Compression

Enable gzip/brotli:

Accept-Encoding: gzip, deflate, br
Content-Encoding: gzip

3. Field Selection

Allow clients to request specific fields:

GET /api/v1/users?fields=id,name,email

GraphQL approach:

{
  users {
    id
    name
    email
  }
}

4. Batch Endpoints

Instead of:

GET /api/v1/users/1
GET /api/v1/users/2
GET /api/v1/users/3

Use:

GET /api/v1/users?ids=1,2,3

Checklist

• [ ] Database indexes on foreign keys
• [ ] Connection pooling configured
• [ ] Response compression enabled
• [ ] Slow query logging
• [ ] API response time monitoring
• [ ] Caching strategy implemented
• [ ] CDN for static assets

---

Summary: Universal Best Practices

1. Security First: HTTPS, authentication, authorization, input validation 2. Consistency: Naming, error formats, response structures 3. Documentation: OpenAPI specs, examples, migration guides 4. Performance: Caching, indexing, pagination, compression 5. Reliability: Error handling, rate limiting, timeouts, retries 6. Observability: Logging, metrics, tracing, monitoring 7. Versioning: Clear strategy, deprecation policy, migration support 8. Developer Experience: Clear docs, predictable behavior, helpful errors

API Design Review Checklist (Production)

Use this checklist for PRDs, RFCs, and PR reviews to keep APIs consistent, secure, and operable.

---

Core

1) Scope and Consumers

• [ ] Audience: public / partner / internal
• [ ] Use case(s) and non-goals listed
• [ ] Backward compatibility expectations documented
• [ ] Data classification (PII/PHI/PCI) documented

2) Resource Model and Semantics

• [ ] Resource names are nouns; actions are minimized and justified
• [ ] HTTP method semantics match intent (GET safe, PUT/DELETE idempotent)
• [ ] Idempotency defined for create/mutation flows (idempotency key or natural key)
• [ ] Pagination is mandatory for list endpoints (cursor preferred for large/real-time lists)
• [ ] Filtering and sorting are explicit, indexed, and bounded (no unbounded scans)
• [ ] Partial responses and field selection strategy decided (if needed for cost control)

3) Versioning and Deprecation

• [ ] Versioning strategy is explicit (URL/header/content negotiation) and consistently applied
• [ ] Deprecation policy exists (notice period, sunset date, comms plan)
• [ ] Breaking-change rules documented (what counts as breaking)
• [ ] Compatibility tests (or contract checks) defined for stable endpoints

4) Error Model (Consistency + Retryability)

• [ ] RFC 9457 Problem Details is the default response format for errors
• [ ] Stable code registry exists (machine-readable, documented)
• [ ] retryable is defined per error code and consistent with HTTP semantics
• [ ] 401 uses WWW-Authenticate where relevant; 429 uses Retry-After where applicable
• [ ] Errors do not leak secrets, stack traces, internal hostnames, or SQL text

Template: template-api-error-model.md

5) AuthN/AuthZ and Multi-Tenancy

• [ ] AuthN mechanism is explicit (OAuth2/OIDC, JWT, mTLS, API keys)
• [ ] AuthZ model is explicit (RBAC/ABAC) with least privilege
• [ ] Tenant boundaries are enforced server-side (never trust tenant ID from client)
• [ ] Audit logging requirements are defined for sensitive operations

6) Rate Limits, Quotas, and Abuse Controls

• [ ] Rate limits are defined per actor (user/org/token) and per route group
• [ ] Limits are observable (rate limit headers, dashboards, alert thresholds)
• [ ] Abuse prevention: replay protection, bot mitigation where needed
• [ ] Traffic shaping plan exists for batch/backfill endpoints

7) Reliability and Operations

• [ ] Timeouts and retries are defined per dependency (client and server)
• [ ] Request IDs and trace propagation are implemented (logs + traces)
• [ ] Metrics defined: latency (p95/p99), error rate, saturation, queue depth
• [ ] Runbook exists for top failure modes (dependency outage, auth outage, overload)
• [ ] Safe deploy plan: canary/gradual rollout, rollback strategy, feature flags if needed

8) Data and Performance

• [ ] N+1 risk addressed (batch endpoints, joins, caching, DataLoader for GraphQL)
• [ ] Queries are bounded and indexed for hot paths
• [ ] Bulk endpoints have throttles and async patterns (202 + job status) if needed
• [ ] Caching strategy is explicit (ETag/If-None-Match, Cache-Control) where appropriate

---

Do / Avoid

Do

• Do define a stable error code taxonomy before the first external client ships
• Do treat rate limits as part of the contract (document and test them)
• Do design for operability: logs/metrics/traces and runbooks are required
• Do require pagination and bounded filters to control cost and reliability

Avoid

• Avoid action endpoints for everything (/doThing); start with resources + verbs
• Avoid breaking changes without an explicit deprecation and migration path
• Avoid returning raw internal errors to clients (leaks + unstable contracts)
• Avoid unbounded list endpoints and expensive filters without quotas

---

Optional: AI/Automation

Use only as a productivity layer; humans own the contract.

• Automated OpenAPI linting (Spectral/Redocly) and diff checks (oasdiff)
• Drafting endpoint tables and examples from an approved design doc
• Generating initial SDK stubs and contract-test scaffolding (human-reviewed)

Bounded Claims

• Generated specs and examples require human review for security and semantics.
• Automation cannot infer business-acceptable risk, quotas, or deprecation timelines.

API Error Model Template (RFC 9457 Problem Details)

Use this template to standardize errors across services and make client behavior predictable.

---

Core

1) Default Response Format

• Content-Type: application/problem+json
• Standard fields: type, title, status, detail, instance
• Extensions (recommended): code, retryable, errors[], trace_id

Example:

{
  "type": "https://api.example.com/problems/validation-error",
  "title": "Validation Error",
  "status": 422,
  "detail": "One or more fields failed validation",
  "instance": "/v1/users",
  "code": "validation_error",
  "retryable": false,
  "trace_id": "01JFDY9W9Q8Y2W7A9JY3VJQ3Z0",
  "errors": [
    { "field": "email", "code": "invalid_format", "message": "Email must be valid" }
  ]
}

2) Error Code Registry (Required)

Rules:

• code is stable (clients can switch on it).
• type is stable and documented (URI per error family).
• title is stable per error type (human-readable).
• detail is instance-specific and safe to expose.

Registry template:

3) HTTP Header Requirements

• 401: WWW-Authenticate (Bearer realm and error details where applicable)
• 429: Retry-After when you can provide a meaningful retry time
• All responses: request correlation header (traceparent or x-request-id) should be returned

4) Validation Error Shape

Guidelines:

• Use errors[] only for field-level issues.
• errors[].code is stable; errors[].message is safe and user-facing.
• Prefer 422 for semantic validation errors; 400 for malformed syntax.

5) Security Requirements

• Do not include stack traces, SQL text, secrets, internal hostnames, or dependency credentials.
• Avoid user enumeration: keep auth errors generic where needed.
• Ensure tenant boundaries: do not leak cross-tenant existence via 404/403 mismatches.

6) Observability Requirements

• Always include a correlation identifier (trace_id and/or traceparent).
• Log the full error server-side with internal diagnostics keyed by trace_id.
• Emit metrics by code and route group to support alerting and SLOs.

---

Do / Avoid

Do

• Do document the code registry and treat it as API surface
• Do keep detail safe and actionable for clients
• Do mark retryability explicitly and keep it consistent with status codes

Avoid

• Avoid inventing a different error shape per endpoint
• Avoid making clients parse human strings to decide behavior
• Avoid leaking sensitive internals in error responses

---

Optional: AI/Automation

• Generate the first draft of code registry entries from existing logs (human-reviewed)
• Summarize incident spikes by code and propose top contributors (human-validated)
• Create contract tests from the registry table (human-owned acceptance)

Bounded Claims

• Automation cannot decide which errors are safe to expose.
• Retryability must be validated against real dependency behavior and SLAs.

API Governance & Design Review Checklist

Production-grade API governance covering deprecation, compatibility, multi-tenancy, and SDK guidelines.

---

API Design Review Checklist

Before Implementation

• [ ] API scope and purpose documented
• [ ] Resource naming follows conventions (nouns, plural)
• [ ] HTTP methods semantically correct (GET=read, POST=create, etc.)
• [ ] Request/response schemas defined in OpenAPI
• [ ] Authentication method specified
• [ ] Rate limits defined
• [ ] Error codes and messages documented
• [ ] Breaking change assessment completed

Resource Design

Request/Response Design

---

Deprecation Policy

Deprecation Timeline

Deprecation Headers

HTTP/1.1 200 OK
Deprecation: Sun, 01 Dec 2025 00:00:00 GMT
Sunset: Sun, 01 Mar 2026 00:00:00 GMT
Link: <https://api.example.com/v2/users>; rel="successor-version"

Deprecation Announcement Template

## API Deprecation Notice: [Endpoint/Version]

**Deprecated**: [Date]
**Sunset Date**: [Date + 90 days minimum]
**Replacement**: [New endpoint or version]

### What's Changing
[Description of deprecated functionality]

### Migration Guide
1. [Step 1]
2. [Step 2]
3. [Step 3]

### Timeline
- [ ] 90 days: Deprecation announced
- [ ] 60 days: Warning logs enabled
- [ ] 30 days: Direct migration outreach
- [ ] 0 days: Endpoint returns 410 Gone

### Support
Contact api-support@example.com for migration assistance.

---

Backward Compatibility Rules

Breaking Changes (Require Major Version)

• Removing an endpoint
• Removing a field from response
• Changing field type (string → number)
• Changing field meaning
• Adding required request field
• Changing authentication method
• Changing error code meanings

Non-Breaking Changes (Safe)

• Adding new endpoint
• Adding optional request field
• Adding response field
• Adding new error codes
• Adding enum values (if client handles unknown gracefully)
• Adding new HTTP methods to existing resource

Compatibility Checklist

• [ ] No fields removed from response
• [ ] No required fields added to request
• [ ] No field types changed
• [ ] Existing error codes unchanged
• [ ] Authentication backward compatible
• [ ] Rate limits not made stricter

Schema Evolution Strategy

# Good: Additive change (non-breaking)
User:
  properties:
    id: string
    name: string
    email: string
    phone: string  # NEW - optional

# Bad: Breaking change
User:
  properties:
    id: string
    fullName: string  # RENAMED from 'name' - BREAKING
    emailAddress: string  # RENAMED from 'email' - BREAKING

---

Multi-Tenant API Patterns

Tenant Isolation Checklist

• [ ] Tenant ID in all database queries
• [ ] Row-level security enabled
• [ ] Cross-tenant data access impossible
• [ ] Tenant context propagated in headers
• [ ] Audit logs include tenant ID
• [ ] Rate limits per tenant

Tenant Identification Methods

Multi-Tenant Request Template

GET /api/v1/users HTTP/1.1
Host: api.example.com
Authorization: Bearer <jwt_with_tenant_claim>
X-Tenant-ID: acme  # Redundant with JWT, but explicit

# Response scoped to tenant
{
  "data": [
    {"id": "user_123", "tenant_id": "acme", "name": "Alice"}
  ],
  "meta": {
    "tenant_id": "acme",
    "total": 1
  }
}

Tenant Data Isolation Patterns

---

Error Model Template

Standard Error Response (RFC 9457, obsoletes RFC 7807)

{
  "type": "https://api.example.com/errors/validation-error",
  "title": "Validation Error",
  "status": 422,
  "detail": "One or more fields failed validation",
  "instance": "/api/v1/users/123",
  "code": "validation_error",
  "retryable": false,
  "trace_id": "abc123def456",
  "errors": [
    {
      "field": "email",
      "code": "invalid_format",
      "message": "Email must be a valid email address"
    },
    {
      "field": "age",
      "code": "out_of_range",
      "message": "Age must be between 0 and 150"
    }
  ]
}

Error Code Registry

Retryability Header

HTTP/1.1 429 Too Many Requests
Retry-After: 60
X-RateLimit-Reset: 1701388800

---

SDK Guidelines

SDK Generation Checklist

• [ ] OpenAPI spec is complete and accurate
• [ ] All endpoints have examples
• [ ] Error responses documented
• [ ] Authentication flows documented
• [ ] Pagination handling documented
• [ ] Rate limit handling documented

SDK Feature Requirements

SDK Error Handling Pattern

// Good SDK error handling
try {
  const user = await client.users.get("user_123");
} catch (error) {
  if (error instanceof ApiError) {
    switch (error.code) {
      case "resource_not_found":
        // Handle 404
        break;
      case "rate_limited":
        // Retry with backoff (SDK should handle)
        break;
      case "authentication_required":
        // Refresh token and retry
        break;
      default:
        // Unknown error
        throw error;
    }
  }
}

---

Do / Avoid

GOOD: Do

• Version all APIs from day one
• Document deprecation policy before first deprecation
• Use semantic versioning for API versions
• Provide migration guides for breaking changes
• Include trace IDs in all error responses
• Test SDK generated code before release
• Monitor API usage before deprecating endpoints

BAD: Avoid

• Removing fields without deprecation period
• Changing field types in existing versions
• Making auth more restrictive without notice
• Deprecating without providing alternatives
• Mixing tenant data without explicit isolation
• Hardcoding error messages (use codes)
• Returning 500 for client errors

---

Anti-Patterns

---

Optional: AI/Automation

Note: These enhance governance but require human oversight.

Automated Governance

• OpenAPI linting in CI/CD (Spectral, Redocly)
• Breaking change detection (oasdiff)
• SDK generation on spec changes
• Contract testing automation (Pact, Dredd)

AI-Assisted Review

• API design suggestions (must be validated)
• Documentation generation (review for accuracy)
• Error message improvement suggestions

Bounded Claims

• AI-generated OpenAPI specs require human review
• Automated deprecation detection needs manual confirmation
• SDK generation requires type verification

---

Related Templates

• api-patterns-universal.md — Cross-framework patterns
• ../error-handling-patterns.md — Error model deep dive
• ../versioning-strategies.md — Version comparison

---

Last Updated: December 2025

Django REST Framework Complete API Template

Production-ready Django REST Framework API with authentication, permissions, serializers, and viewsets.

Project Setup

# Create project
django-admin startproject api_project
cd api_project
django-admin startapp users

# Install dependencies
pip install djangorestframework djangorestframework-simplejwt django-filter drf-spectacular python-dotenv psycopg2-binary redis django-redis django-cors-headers

Requirements (requirements.txt)

Django==5.0
djangorestframework==3.14.0
djangorestframework-simplejwt==5.3.1
django-filter==23.5
drf-spectacular==0.27.0
python-dotenv==1.0.0
psycopg2-binary==2.9.9
redis==5.0.1
django-redis==5.4.0
django-cors-headers==4.3.1
gunicorn==21.2.0

Settings (api_project/settings.py)

import os
from pathlib import Path
from datetime import timedelta
from dotenv import load_dotenv

load_dotenv()

BASE_DIR = Path(__file__).resolve().parent.parent
SECRET_KEY = os.getenv('SECRET_KEY')
DEBUG = os.getenv('DEBUG', 'False') == 'True'
ALLOWED_HOSTS = os.getenv('ALLOWED_HOSTS', 'localhost').split(',')

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    # Third party
    'rest_framework',
    'rest_framework_simplejwt',
    'django_filters',
    'drf_spectacular',
    'corsheaders',
    # Local apps
    'users',
]

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'corsheaders.middleware.CorsMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
]

ROOT_URLCONF = 'api_project.urls'

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': os.getenv('DB_NAME'),
        'USER': os.getenv('DB_USER'),
        'PASSWORD': os.getenv('DB_PASSWORD'),
        'HOST': os.getenv('DB_HOST', 'localhost'),
        'PORT': os.getenv('DB_PORT', '5432'),
    }
}

# REST Framework
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework_simplejwt.authentication.JWTAuthentication',
    ],
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ],
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.LimitOffsetPagination',
    'PAGE_SIZE': 20,
    'DEFAULT_FILTER_BACKENDS': [
        'django_filters.rest_framework.DjangoFilterBackend',
        'rest_framework.filters.OrderingFilter',
        'rest_framework.filters.SearchFilter',
    ],
    'DEFAULT_SCHEMA_CLASS': 'drf_spectacular.openapi.AutoSchema',
    'EXCEPTION_HANDLER': 'api_project.exceptions.custom_exception_handler',
}

# JWT Settings
SIMPLE_JWT = {
    'ACCESS_TOKEN_LIFETIME': timedelta(minutes=15),
    'REFRESH_TOKEN_LIFETIME': timedelta(days=7),
    'ROTATE_REFRESH_TOKENS': True,
    'BLACKLIST_AFTER_ROTATION': True,
    'ALGORITHM': 'HS256',
    'SIGNING_KEY': SECRET_KEY,
    'AUTH_HEADER_TYPES': ('Bearer',),
}

# CORS
CORS_ALLOWED_ORIGINS = os.getenv('CORS_ORIGINS', 'http://localhost:3000').split(',')

# OpenAPI Documentation
SPECTACULAR_SETTINGS = {
    'TITLE': 'API Documentation',
    'VERSION': '1.0.0',
    'SERVE_INCLUDE_SCHEMA': False,
}

# Cache (Redis)
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': os.getenv('REDIS_URL', 'redis://127.0.0.1:6379/1'),
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
        }
    }
}

AUTH_USER_MODEL = 'users.User'

Custom User Model (users/models.py)

from django.contrib.auth.models import AbstractBaseUser, BaseUserManager, PermissionsMixin
from django.db import models
import uuid

class UserManager(BaseUserManager):
    def create_user(self, email, password=None, **extra_fields):
        if not email:
            raise ValueError('Email is required')
        email = self.normalize_email(email)
        user = self.model(email=email, **extra_fields)
        user.set_password(password)
        user.save(using=self._db)
        return user

    def create_superuser(self, email, password=None, **extra_fields):
        extra_fields.setdefault('is_staff', True)
        extra_fields.setdefault('is_superuser', True)
        extra_fields.setdefault('is_admin', True)
        return self.create_user(email, password, **extra_fields)

class User(AbstractBaseUser, PermissionsMixin):
    class Status(models.TextChoices):
        ACTIVE = 'ACTIVE', 'Active'
        INACTIVE = 'INACTIVE', 'Inactive'
        SUSPENDED = 'SUSPENDED', 'Suspended'

    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    email = models.EmailField(unique=True, db_index=True)
    full_name = models.CharField(max_length=100)
    status = models.CharField(max_length=20, choices=Status.choices, default=Status.ACTIVE)
    is_admin = models.BooleanField(default=False)
    is_staff = models.BooleanField(default=False)
    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    objects = UserManager()

    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ['full_name']

    class Meta:
        db_table = 'users'
        ordering = ['-created_at']

    def __str__(self):
        return self.email

Serializers (users/serializers.py)

from rest_framework import serializers
from .models import User

class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'email', 'full_name', 'status', 'is_admin', 'created_at', 'updated_at']
        read_only_fields = ['id', 'created_at', 'updated_at']

class UserCreateSerializer(serializers.ModelSerializer):
    password = serializers.CharField(write_only=True, min_length=8, max_length=100)

    class Meta:
        model = User
        fields = ['email', 'full_name', 'password']

    def create(self, validated_data):
        return User.objects.create_user(**validated_data)

class UserUpdateSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['full_name', 'status']

class LoginSerializer(serializers.Serializer):
    email = serializers.EmailField()
    password = serializers.CharField(write_only=True)

Permissions (users/permissions.py)

from rest_framework import permissions

class IsAdminUser(permissions.BasePermission):
    """Allow access only to admin users."""
    def has_permission(self, request, view):
        return request.user and request.user.is_authenticated and request.user.is_admin

class IsOwnerOrAdmin(permissions.BasePermission):
    """Allow access to resource owner or admin."""
    def has_object_permission(self, request, view, obj):
        return obj == request.user or request.user.is_admin

ViewSets (users/views.py)

from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated, AllowAny
from rest_framework_simplejwt.tokens import RefreshToken
from django.contrib.auth import authenticate
from .models import User
from .serializers import (
    UserSerializer,
    UserCreateSerializer,
    UserUpdateSerializer,
    LoginSerializer
)
from .permissions import IsAdminUser, IsOwnerOrAdmin

class AuthViewSet(viewsets.GenericViewSet):
    """Authentication endpoints."""
    permission_classes = [AllowAny]
    serializer_class = LoginSerializer

    @action(detail=False, methods=['post'])
    def login(self, request):
        """Authenticate user and return JWT tokens."""
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        user = authenticate(
            email=serializer.validated_data['email'],
            password=serializer.validated_data['password']
        )

        if not user:
            return Response(
                {'detail': 'Invalid credentials'},
                status=status.HTTP_401_UNAUTHORIZED
            )

        refresh = RefreshToken.for_user(user)
        return Response({
            'access_token': str(refresh.access_token),
            'refresh_token': str(refresh),
            'token_type': 'Bearer',
            'expires_in': 900  # 15 minutes
        })

class UserViewSet(viewsets.ModelViewSet):
    """User management endpoints."""
    queryset = User.objects.all()
    serializer_class = UserSerializer
    filterset_fields = ['status', 'is_admin']
    search_fields = ['email', 'full_name']
    ordering_fields = ['created_at', 'email']

    def get_serializer_class(self):
        if self.action == 'create':
            return UserCreateSerializer
        elif self.action in ['update', 'partial_update']:
            return UserUpdateSerializer
        return UserSerializer

    def get_permissions(self):
        if self.action == 'create':
            return [IsAdminUser()]
        elif self.action in ['update', 'partial_update', 'destroy']:
            return [IsAuthenticated(), IsOwnerOrAdmin()]
        return [IsAuthenticated()]

    @action(detail=False, methods=['get'])
    def me(self, request):
        """Get current user profile."""
        serializer = self.get_serializer(request.user)
        return Response(serializer.data)

URL Configuration (users/urls.py)

from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import AuthViewSet, UserViewSet

router = DefaultRouter()
router.register(r'auth', AuthViewSet, basename='auth')
router.register(r'users', UserViewSet, basename='users')

urlpatterns = [
    path('', include(router.urls)),
]

Main URLs (api_project/urls.py)

from django.contrib import admin
from django.urls import path, include
from drf_spectacular.views import SpectacularAPIView, SpectacularSwaggerView

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/v1/', include('users.urls')),
    path('api/schema/', SpectacularAPIView.as_view(), name='schema'),
    path('api/docs/', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui'),
]

Custom Exception Handler (api_project/exceptions.py)

from rest_framework.views import exception_handler
from rest_framework.response import Response

def custom_exception_handler(exc, context):
    """RFC 9457 (Problem Details) format."""
    response = exception_handler(exc, context)

    if response is not None:
        request = context.get('request')
        custom_response = {
            'type': f'https://api.example.com/errors/{response.status_code}',
            'title': exc.__class__.__name__,
            'status': response.status_code,
            'detail': str(exc),
            'instance': request.path if request else None,
        }

        if isinstance(response.data, dict):
            custom_response['errors'] = response.data

        response.data = custom_response

    return response

Environment Variables (.env)

SECRET_KEY=your-secret-key-change-in-production
DEBUG=True
ALLOWED_HOSTS=localhost,127.0.0.1

DB_NAME=api_db
DB_USER=postgres
DB_PASSWORD=password
DB_HOST=localhost
DB_PORT=5432

REDIS_URL=redis://127.0.0.1:6379/1

CORS_ORIGINS=http://localhost:3000,http://localhost:8080

Running the Application

# Migrate database
python manage.py makemigrations
python manage.py migrate

# Create superuser
python manage.py createsuperuser

# Run development server
python manage.py runserver

# API docs at:
# - Swagger: http://localhost:8000/api/docs/
# - Schema: http://localhost:8000/api/schema/

Key Features

[check] Django REST Framework: Full-featured REST API [check] JWT Authentication: Simple JWT with refresh tokens [check] Custom User Model: UUID primary key, flexible fields [check] Permissions: RBAC with custom permissions [check] Serializers: Validation and data transformation [check] ViewSets: DRY API endpoints [check] Filtering: Search, filter, ordering built-in [check] Pagination: Limit/offset pagination [check] OpenAPI: Auto-generated Swagger docs [check] Error Handling: RFC 9457 (Problem Details) [check] CORS: Cross-origin support [check] Redis: Caching ready

Best Practices Applied

• Custom user model with UUID
• ViewSet and serializer patterns
• Permission-based access control
• Environment-based configuration
• Database migrations
• OpenAPI documentation
• Standardized error responses
• Django ORM best practices

Express.js Complete API Template

Production-ready Express.js/TypeScript API with authentication, database, validation, error handling, and OpenAPI docs.

Project Structure

express-api/
├── src/
│   ├── index.ts              # Application entry point
│   ├── app.ts                # Express app configuration
│   ├── config/
│   │   └── index.ts          # Environment configuration
│   ├── db/
│   │   ├── client.ts         # Database connection (Prisma/TypeORM)
│   │   └── migrations/
│   ├── middleware/
│   │   ├── auth.ts           # JWT authentication
│   │   ├── errorHandler.ts  # Global error handling
│   │   ├── validate.ts       # Request validation
│   │   └── rateLimit.ts      # Rate limiting
│   ├── models/               # Database models
│   │   └── User.ts
│   ├── routes/
│   │   ├── index.ts
│   │   ├── auth.routes.ts
│   │   └── users.routes.ts
│   ├── controllers/
│   │   ├── auth.controller.ts
│   │   └── users.controller.ts
│   ├── services/
│   │   └── user.service.ts   # Business logic
│   ├── validators/
│   │   └── user.validator.ts # Zod schemas
│   └── utils/
│       ├── jwt.ts            # Token utilities
│       ├── password.ts       # Hashing utilities
│       └── pagination.ts     # Pagination helpers
├── tests/
│   ├── integration/
│   └── unit/
├── prisma/
│   └── schema.prisma         # Database schema
├── .env.example
├── package.json
├── tsconfig.json
└── openapi.yaml              # API documentation

Centralization Guide

Important: The code patterns in this template should be extracted to src/utils/. Do not duplicate these utilities across controllers/services.

Pattern: Create utilities once in src/utils/, import everywhere via:

import { hashPassword, verifyToken } from '@/utils/auth';
import { ApiError, NotFoundError } from '@/utils/errors';

1. Dependencies (package.json)

{
  "name": "express-api",
  "version": "1.0.0",
  "scripts": {
    "dev": "tsx watch src/index.ts",
    "build": "tsc",
    "start": "node dist/index.js",
    "test": "jest",
    "migrate": "prisma migrate dev",
    "generate": "prisma generate"
  },
  "dependencies": {
    "express": "^4.18.2",
    "express-async-errors": "^3.1.1",
    "@prisma/client": "^5.8.0",
    "zod": "^3.22.4",
    "bcrypt": "^5.1.1",
    "jsonwebtoken": "^9.0.2",
    "dotenv": "^16.3.1",
    "helmet": "^7.1.0",
    "cors": "^2.8.5",
    "express-rate-limit": "^7.1.5",
    "redis": "^4.6.12",
    "morgan": "^1.10.0",
    "winston": "^3.11.0"
  },
  "devDependencies": {
    "@types/express": "^4.17.21",
    "@types/node": "^20.10.6",
    "@types/bcrypt": "^5.0.2",
    "@types/jsonwebtoken": "^9.0.5",
    "@types/cors": "^2.8.17",
    "@types/morgan": "^1.9.9",
    "typescript": "^5.3.3",
    "tsx": "^4.7.0",
    "jest": "^29.7.0",
    "@types/jest": "^29.5.11",
    "supertest": "^6.3.3",
    "@types/supertest": "^6.0.2",
    "prisma": "^5.8.0"
  }
}

2. TypeScript Configuration (tsconfig.json)

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "commonjs",
    "lib": ["ES2022"],
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true,
    "resolveJsonModule": true,
    "moduleResolution": "node",
    "declaration": true,
    "sourceMap": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules", "dist", "tests"]
}

3. Configuration (src/config/index.ts)

import dotenv from 'dotenv';
import { z } from 'zod';

dotenv.config();

const envSchema = z.object({
  NODE_ENV: z.enum(['development', 'production', 'test']).default('development'),
  PORT: z.string().default('3000'),
  DATABASE_URL: z.string(),
  JWT_SECRET: z.string(),
  JWT_ACCESS_EXPIRE: z.string().default('15m'),
  JWT_REFRESH_EXPIRE: z.string().default('7d'),
  REDIS_URL: z.string().optional(),
  CORS_ORIGINS: z.string().default('http://localhost:3000'),
  RATE_LIMIT_WINDOW_MS: z.string().default('60000'),
  RATE_LIMIT_MAX_REQUESTS: z.string().default('100'),
});

const parsed = envSchema.safeParse(process.env);

if (!parsed.success) {
  console.error('Invalid environment variables:', parsed.error.flatten().fieldErrors);
  process.exit(1);
}

export const config = {
  env: parsed.data.NODE_ENV,
  port: parseInt(parsed.data.PORT),
  database: {
    url: parsed.data.DATABASE_URL,
  },
  jwt: {
    secret: parsed.data.JWT_SECRET,
    accessExpire: parsed.data.JWT_ACCESS_EXPIRE,
    refreshExpire: parsed.data.JWT_REFRESH_EXPIRE,
  },
  redis: {
    url: parsed.data.REDIS_URL,
  },
  cors: {
    origins: parsed.data.CORS_ORIGINS.split(','),
  },
  rateLimit: {
    windowMs: parseInt(parsed.data.RATE_LIMIT_WINDOW_MS),
    maxRequests: parseInt(parsed.data.RATE_LIMIT_MAX_REQUESTS),
  },
} as const;

4. Database Client (src/db/client.ts)

import { PrismaClient } from '@prisma/client';
import { config } from '../config';

const prisma = new PrismaClient({
  log: config.env === 'development' ? ['query', 'error', 'warn'] : ['error'],
});

// Graceful shutdown
process.on('beforeExit', async () => {
  await prisma.$disconnect();
});

export { prisma };

5. Prisma Schema (prisma/schema.prisma)

datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

generator client {
  provider = "prisma-client-js"
}

enum UserStatus {
  ACTIVE
  INACTIVE
  SUSPENDED
}

model User {
  id        String     @id @default(uuid())
  email     String     @unique
  password  String
  fullName  String     @map("full_name")
  status    UserStatus @default(ACTIVE)
  isAdmin   Boolean    @default(false) @map("is_admin")
  createdAt DateTime   @default(now()) @map("created_at")
  updatedAt DateTime   @updatedAt @map("updated_at")

  @@map("users")
  @@index([email])
}

6. Request Validators (src/validators/user.validator.ts)

import { z } from 'zod';

export const createUserSchema = z.object({
  body: z.object({
    email: z.string().email(),
    fullName: z.string().min(2).max(100),
    password: z.string().min(8).max(100),
  }),
});

export const updateUserSchema = z.object({
  body: z.object({
    fullName: z.string().min(2).max(100).optional(),
    status: z.enum(['ACTIVE', 'INACTIVE', 'SUSPENDED']).optional(),
  }),
  params: z.object({
    id: z.string().uuid(),
  }),
});

export const getUserSchema = z.object({
  params: z.object({
    id: z.string().uuid(),
  }),
});

export const listUsersSchema = z.object({
  query: z.object({
    limit: z.string().transform(Number).pipe(z.number().min(1).max(100)).default('20'),
    offset: z.string().transform(Number).pipe(z.number().min(0)).default('0'),
    sort: z.string().optional(),
  }),
});

export const loginSchema = z.object({
  body: z.object({
    email: z.string().email(),
    password: z.string().min(1),
  }),
});

export type CreateUserInput = z.infer<typeof createUserSchema>;
export type UpdateUserInput = z.infer<typeof updateUserSchema>;
export type GetUserInput = z.infer<typeof getUserSchema>;
export type ListUsersInput = z.infer<typeof listUsersSchema>;
export type LoginInput = z.infer<typeof loginSchema>;

7. Validation Middleware (src/middleware/validate.ts)

import { Request, Response, NextFunction } from 'express';
import { ZodSchema, ZodError } from 'zod';
import { ApiError } from '../utils/errors';

export const validate = (schema: ZodSchema) => {
  return async (req: Request, res: Response, next: NextFunction) => {
    try {
      await schema.parseAsync({
        body: req.body,
        query: req.query,
        params: req.params,
      });
      next();
    } catch (error) {
      if (error instanceof ZodError) {
        const errors = error.errors.map((e) => ({
          field: e.path.join('.'),
          message: e.message,
          code: e.code,
        }));

        next(
          new ApiError(422, 'Validation Error', {
            errors,
            type: 'https://api.example.com/errors/validation',
          })
        );
      } else {
        next(error);
      }
    }
  };
};

8. JWT Utilities (src/utils/jwt.ts)

import jwt from 'jsonwebtoken';
import { config } from '../config';

export interface JwtPayload {
  userId: string;
  email: string;
  isAdmin: boolean;
}

export const createAccessToken = (payload: JwtPayload): string => {
  return jwt.sign(payload, config.jwt.secret, {
    expiresIn: config.jwt.accessExpire,
  });
};

export const createRefreshToken = (payload: JwtPayload): string => {
  return jwt.sign(payload, config.jwt.secret, {
    expiresIn: config.jwt.refreshExpire,
  });
};

export const verifyToken = (token: string): JwtPayload => {
  try {
    return jwt.verify(token, config.jwt.secret) as JwtPayload;
  } catch (error) {
    throw new Error('Invalid token');
  }
};

9. Password Utilities (src/utils/password.ts)

import bcrypt from 'bcrypt';

const SALT_ROUNDS = 10;

export const hashPassword = async (password: string): Promise<string> => {
  return bcrypt.hash(password, SALT_ROUNDS);
};

export const comparePassword = async (
  password: string,
  hash: string
): Promise<boolean> => {
  return bcrypt.compare(password, hash);
};

10. Auth Middleware (src/middleware/auth.ts)

import { Request, Response, NextFunction } from 'express';
import { verifyToken, JwtPayload } from '../utils/jwt';
import { ApiError } from '../utils/errors';

declare global {
  namespace Express {
    interface Request {
      user?: JwtPayload;
    }
  }
}

export const authenticate = (req: Request, res: Response, next: NextFunction) => {
  try {
    const authHeader = req.headers.authorization;
    if (!authHeader || !authHeader.startsWith('Bearer ')) {
      throw new ApiError(401, 'No token provided');
    }

    const token = authHeader.substring(7);
    const payload = verifyToken(token);

    req.user = payload;
    next();
  } catch (error) {
    next(new ApiError(401, 'Invalid or expired token'));
  }
};

export const requireAdmin = (req: Request, res: Response, next: NextFunction) => {
  if (!req.user?.isAdmin) {
    return next(new ApiError(403, 'Admin privileges required'));
  }
  next();
};

11. Error Handler (src/middleware/errorHandler.ts)

import { Request, Response, NextFunction } from 'express';
import { config } from '../config';

export class ApiError extends Error {
  constructor(
    public statusCode: number,
    public message: string,
    public details?: any
  ) {
    super(message);
    this.name = 'ApiError';
  }
}

export const errorHandler = (
  error: Error | ApiError,
  req: Request,
  res: Response,
  next: NextFunction
) => {
  if (error instanceof ApiError) {
    return res.status(error.statusCode).json({
      type: error.details?.type || `https://api.example.com/errors/${error.statusCode}`,
      title: error.message,
      status: error.statusCode,
      detail: error.message,
      instance: req.path,
      ...(error.details && { errors: error.details.errors }),
      ...(config.env === 'development' && { stack: error.stack }),
    });
  }

  // Unhandled errors
  console.error('Unhandled error:', error);
  res.status(500).json({
    type: 'https://api.example.com/errors/500',
    title: 'Internal Server Error',
    status: 500,
    detail: config.env === 'production' ? 'An unexpected error occurred' : error.message,
    instance: req.path,
    ...(config.env === 'development' && { stack: error.stack }),
  });
};

export const notFoundHandler = (req: Request, res: Response) => {
  res.status(404).json({
    type: 'https://api.example.com/errors/404',
    title: 'Not Found',
    status: 404,
    detail: `Route ${req.method} ${req.path} not found`,
    instance: req.path,
  });
};

12. Rate Limiting (src/middleware/rateLimit.ts)

import rateLimit from 'express-rate-limit';
import { config } from '../config';

export const rateLimiter = rateLimit({
  windowMs: config.rateLimit.windowMs,
  max: config.rateLimit.maxRequests,
  message: {
    type: 'https://api.example.com/errors/429',
    title: 'Too Many Requests',
    status: 429,
    detail: 'Rate limit exceeded',
  },
  standardHeaders: true,
  legacyHeaders: false,
});

13. User Service (src/services/user.service.ts)

import { prisma } from '../db/client';
import { hashPassword } from '../utils/password';
import { ApiError } from '../middleware/errorHandler';
import { User, UserStatus } from '@prisma/client';

export class UserService {
  async createUser(data: {
    email: string;
    password: string;
    fullName: string;
  }): Promise<Omit<User, 'password'>> {
    const existing = await prisma.user.findUnique({
      where: { email: data.email },
    });

    if (existing) {
      throw new ApiError(409, 'Email already registered');
    }

    const hashedPassword = await hashPassword(data.password);

    const user = await prisma.user.create({
      data: {
        email: data.email,
        password: hashedPassword,
        fullName: data.fullName,
      },
      select: {
        id: true,
        email: true,
        fullName: true,
        status: true,
        isAdmin: true,
        createdAt: true,
        updatedAt: true,
      },
    });

    return user;
  }

  async getUserById(id: string): Promise<Omit<User, 'password'> | null> {
    return prisma.user.findUnique({
      where: { id },
      select: {
        id: true,
        email: true,
        fullName: true,
        status: true,
        isAdmin: true,
        createdAt: true,
        updatedAt: true,
      },
    });
  }

  async getUserByEmail(email: string): Promise<User | null> {
    return prisma.user.findUnique({
      where: { email },
    });
  }

  async listUsers(limit: number, offset: number) {
    const [users, total] = await Promise.all([
      prisma.user.findMany({
        take: limit,
        skip: offset,
        orderBy: { createdAt: 'desc' },
        select: {
          id: true,
          email: true,
          fullName: true,
          status: true,
          isAdmin: true,
          createdAt: true,
          updatedAt: true,
        },
      }),
      prisma.user.count(),
    ]);

    return { users, total };
  }

  async updateUser(
    id: string,
    data: { fullName?: string; status?: UserStatus }
  ): Promise<Omit<User, 'password'>> {
    const user = await prisma.user.update({
      where: { id },
      data,
      select: {
        id: true,
        email: true,
        fullName: true,
        status: true,
        isAdmin: true,
        createdAt: true,
        updatedAt: true,
      },
    });

    return user;
  }

  async deleteUser(id: string): Promise<void> {
    await prisma.user.delete({
      where: { id },
    });
  }
}

export const userService = new UserService();

14. Auth Controller (src/controllers/auth.controller.ts)

import { Request, Response, NextFunction } from 'express';
import { userService } from '../services/user.service';
import { comparePassword } from '../utils/password';
import { createAccessToken, createRefreshToken } from '../utils/jwt';
import { ApiError } from '../middleware/errorHandler';

export class AuthController {
  async login(req: Request, res: Response, next: NextFunction) {
    try {
      const { email, password } = req.body;

      const user = await userService.getUserByEmail(email);
      if (!user || !(await comparePassword(password, user.password))) {
        throw new ApiError(401, 'Invalid email or password');
      }

      const payload = {
        userId: user.id,
        email: user.email,
        isAdmin: user.isAdmin,
      };

      const accessToken = createAccessToken(payload);
      const refreshToken = createRefreshToken(payload);

      res.json({
        accessToken,
        refreshToken,
        tokenType: 'Bearer',
        expiresIn: 900, // 15 minutes
      });
    } catch (error) {
      next(error);
    }
  }
}

export const authController = new AuthController();

15. Users Controller (src/controllers/users.controller.ts)

import { Request, Response, NextFunction } from 'express';
import { userService } from '../services/user.service';
import { ApiError } from '../middleware/errorHandler';

export class UsersController {
  async create(req: Request, res: Response, next: NextFunction) {
    try {
      const user = await userService.createUser(req.body);
      res.status(201).json(user);
    } catch (error) {
      next(error);
    }
  }

  async list(req: Request, res: Response, next: NextFunction) {
    try {
      const { limit, offset } = req.query as { limit: string; offset: string };
      const { users, total } = await userService.listUsers(
        parseInt(limit),
        parseInt(offset)
      );

      res.json({
        data: users,
        meta: {
          total,
          limit: parseInt(limit),
          offset: parseInt(offset),
          hasMore: parseInt(offset) + parseInt(limit) < total,
        },
      });
    } catch (error) {
      next(error);
    }
  }

  async getById(req: Request, res: Response, next: NextFunction) {
    try {
      const user = await userService.getUserById(req.params.id);
      if (!user) {
        throw new ApiError(404, 'User not found');
      }
      res.json(user);
    } catch (error) {
      next(error);
    }
  }

  async getCurrent(req: Request, res: Response, next: NextFunction) {
    try {
      const user = await userService.getUserById(req.user!.userId);
      if (!user) {
        throw new ApiError(404, 'User not found');
      }
      res.json(user);
    } catch (error) {
      next(error);
    }
  }

  async update(req: Request, res: Response, next: NextFunction) {
    try {
      // Authorization check
      if (req.params.id !== req.user!.userId && !req.user!.isAdmin) {
        throw new ApiError(403, 'Not authorized to update this user');
      }

      const user = await userService.updateUser(req.params.id, req.body);
      res.json(user);
    } catch (error) {
      next(error);
    }
  }

  async delete(req: Request, res: Response, next: NextFunction) {
    try {
      await userService.deleteUser(req.params.id);
      res.status(204).send();
    } catch (error) {
      next(error);
    }
  }
}

export const usersController = new UsersController();

16. Routes (src/routes/users.routes.ts)

import { Router } from 'express';
import { usersController } from '../controllers/users.controller';
import { authenticate, requireAdmin } from '../middleware/auth';
import { validate } from '../middleware/validate';
import {
  createUserSchema,
  updateUserSchema,
  getUserSchema,
  listUsersSchema,
} from '../validators/user.validator';

const router = Router();

router.post(
  '/',
  authenticate,
  requireAdmin,
  validate(createUserSchema),
  usersController.create
);

router.get('/', authenticate, validate(listUsersSchema), usersController.list);

router.get('/me', authenticate, usersController.getCurrent);

router.get(
  '/:id',
  authenticate,
  validate(getUserSchema),
  usersController.getById
);

router.patch(
  '/:id',
  authenticate,
  validate(updateUserSchema),
  usersController.update
);

router.delete(
  '/:id',
  authenticate,
  requireAdmin,
  validate(getUserSchema),
  usersController.delete
);

export default router;

17. Express App (src/app.ts)

import express from 'express';
import 'express-async-errors';
import helmet from 'helmet';
import cors from 'cors';
import morgan from 'morgan';
import { config } from './config';
import { errorHandler, notFoundHandler } from './middleware/errorHandler';
import { rateLimiter } from './middleware/rateLimit';
import authRoutes from './routes/auth.routes';
import usersRoutes from './routes/users.routes';

const app = express();

// Security middleware
app.use(helmet());
app.use(cors({ origin: config.cors.origins, credentials: true }));
app.use(rateLimiter);

// Logging
app.use(morgan(config.env === 'production' ? 'combined' : 'dev'));

// Body parsing
app.use(express.json());
app.use(express.urlencoded({ extended: true }));

// Health check
app.get('/health', (req, res) => {
  res.json({ status: 'healthy', timestamp: new Date().toISOString() });
});

// API routes
app.use('/api/v1/auth', authRoutes);
app.use('/api/v1/users', usersRoutes);

// Error handling (must be last)
app.use(notFoundHandler);
app.use(errorHandler);

export { app };

18. Server Entry Point (src/index.ts)

import { app } from './app';
import { config } from './config';
import { prisma } from './db/client';

const server = app.listen(config.port, () => {
  console.log(`Server running on port ${config.port}`);
  console.log(`Environment: ${config.env}`);
});

// Graceful shutdown
const gracefulShutdown = async () => {
  console.log('Shutting down gracefully...');

  server.close(async () => {
    await prisma.$disconnect();
    process.exit(0);
  });

  setTimeout(() => {
    console.error('Forced shutdown');
    process.exit(1);
  }, 10000);
};

process.on('SIGTERM', gracefulShutdown);
process.on('SIGINT', gracefulShutdown);

19. Environment Variables (.env.example)

NODE_ENV=development
PORT=3000

DATABASE_URL="postgresql://user:password@localhost:5432/dbname"

JWT_SECRET="your-secret-key-change-in-production"
JWT_ACCESS_EXPIRE="15m"
JWT_REFRESH_EXPIRE="7d"

REDIS_URL="redis://localhost:6379"

CORS_ORIGINS="http://localhost:3000,http://localhost:8080"

RATE_LIMIT_WINDOW_MS=60000
RATE_LIMIT_MAX_REQUESTS=100

Running the Application

# Install dependencies
npm install

# Generate Prisma client
npm run generate

# Run migrations
npm run migrate

# Development
npm run dev

# Production build
npm run build
npm start

# Testing
npm test

Key Features

[check] TypeScript: Full type safety [check] Zod: Runtime validation [check] Prisma: Type-safe ORM [check] JWT: Authentication with access/refresh tokens [check] RBAC: Role-based authorization [check] Error handling: RFC 9457 (Problem Details) [check] Rate limiting: Express rate limit [check] Security: Helmet, CORS, bcrypt [check] Logging: Morgan + Winston ready [check] Testing: Jest + Supertest setup [check] Graceful shutdown: Proper cleanup

Best Practices Applied

• Controller-Service-Repository pattern
• Request validation with Zod
• Centralized error handling
• JWT token security
• Database connection pooling
• Environment-based configuration
• Structured logging
• Graceful shutdown handling
• Type safety throughout
• Async error handling with express-async-errors

FastAPI Complete API Template

Production-ready FastAPI application with authentication, database, validation, error handling, and OpenAPI docs.

Project Structure

fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py              # Application entry point
│   ├── config.py            # Configuration management
│   ├── database.py          # Database connection
│   ├── dependencies.py      # Dependency injection
│   ├── models/              # SQLAlchemy models
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/             # Pydantic schemas
│   │   ├── __init__.py
│   │   └── user.py
│   ├── routers/             # API endpoints
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   └── users.py
│   ├── services/            # Business logic
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── utils/
│       ├── __init__.py
│       ├── security.py      # JWT, password hashing
│       └── pagination.py    # Pagination helpers
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_users.py
├── alembic/                 # Database migrations
├── .env.example
├── requirements.txt
└── pyproject.toml

1. Dependencies (requirements.txt)

fastapi==0.109.0
uvicorn[standard]==0.27.0
sqlalchemy==2.0.25
alembic==1.13.1
pydantic==2.5.3
pydantic-settings==2.1.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
psycopg2-binary==2.9.9      # PostgreSQL
asyncpg==0.29.0             # Async PostgreSQL
redis==5.0.1
python-dotenv==1.0.0
pytest==7.4.4
pytest-asyncio==0.23.3
httpx==0.26.0               # For testing

2. Configuration (app/config.py)

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    """Application settings with environment variable support."""

    # App
    APP_NAME: str = "FastAPI App"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = False

    # Database
    DATABASE_URL: str = "postgresql://user:pass@localhost/dbname"
    ASYNC_DATABASE_URL: str = "postgresql+asyncpg://user:pass@localhost/dbname"

    # Security
    SECRET_KEY: str  # Required, must be in .env
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7

    # CORS
    CORS_ORIGINS: list[str] = ["http://localhost:3000"]

    # Rate Limiting
    RATE_LIMIT_PER_MINUTE: int = 60

    # Pagination
    DEFAULT_PAGE_SIZE: int = 20
    MAX_PAGE_SIZE: int = 100

    model_config = SettingsConfigDict(
        env_file=".env",
        case_sensitive=True
    )

@lru_cache()
def get_settings() -> Settings:
    """Cached settings instance."""
    return Settings()

3. Database Setup (app/database.py)

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.config import get_settings

settings = get_settings()

# Sync engine (for migrations)
engine = create_engine(
    settings.DATABASE_URL,
    pool_pre_ping=True,
    pool_size=5,
    max_overflow=10
)

# Async engine (for API operations)
async_engine = create_async_engine(
    settings.ASYNC_DATABASE_URL,
    pool_pre_ping=True,
    pool_size=5,
    max_overflow=10,
    echo=settings.DEBUG
)

AsyncSessionLocal = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)

Base = declarative_base()

async def get_db():
    """Dependency for database sessions."""
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

4. Models (app/models/user.py)

from sqlalchemy import Column, String, Boolean, DateTime, Enum
from sqlalchemy.dialects.postgresql import UUID
from datetime import datetime
import uuid
import enum
from app.database import Base

class UserStatus(str, enum.Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    SUSPENDED = "suspended"

class User(Base):
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email = Column(String(255), unique=True, nullable=False, index=True)
    hashed_password = Column(String(255), nullable=False)
    full_name = Column(String(100), nullable=False)
    status = Column(Enum(UserStatus), default=UserStatus.ACTIVE, nullable=False)
    is_admin = Column(Boolean, default=False, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)

5. Schemas (app/schemas/user.py)

from pydantic import BaseModel, EmailStr, Field, ConfigDict
from datetime import datetime
from uuid import UUID
from typing import Optional

class UserStatus(str):
    ACTIVE = "active"
    INACTIVE = "inactive"
    SUSPENDED = "suspended"

class UserBase(BaseModel):
    """Base user schema with common fields."""
    email: EmailStr
    full_name: str = Field(..., min_length=2, max_length=100)

class UserCreate(UserBase):
    """Schema for user creation."""
    password: str = Field(..., min_length=8, max_length=100)

class UserUpdate(BaseModel):
    """Schema for user updates (all fields optional)."""
    full_name: Optional[str] = Field(None, min_length=2, max_length=100)
    status: Optional[UserStatus] = None

class UserInDB(UserBase):
    """User schema with database fields."""
    id: UUID
    status: UserStatus
    is_admin: bool
    created_at: datetime
    updated_at: datetime

    model_config = ConfigDict(from_attributes=True)

class UserResponse(UserInDB):
    """Public user response (no sensitive data)."""
    pass

class Token(BaseModel):
    """JWT token response."""
    access_token: str
    refresh_token: str
    token_type: str = "bearer"
    expires_in: int

class TokenData(BaseModel):
    """Decoded token payload."""
    user_id: UUID
    email: str
    is_admin: bool

6. Security Utils (app/utils/security.py)

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import HTTPException, status, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from app.config import get_settings
from app.schemas.user import TokenData
from uuid import UUID

settings = get_settings()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
security = HTTPBearer()

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify password against hash."""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """Hash password."""
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """Create JWT access token."""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)

    to_encode.update({"exp": expire, "type": "access"})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

def create_refresh_token(data: dict) -> str:
    """Create JWT refresh token."""
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire, "type": "refresh"})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

def decode_token(token: str) -> TokenData:
    """Decode and validate JWT token."""
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        user_id: str = payload.get("sub")
        email: str = payload.get("email")
        is_admin: bool = payload.get("is_admin", False)

        if user_id is None or email is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid token"
            )

        return TokenData(user_id=UUID(user_id), email=email, is_admin=is_admin)

    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> TokenData:
    """Dependency to get current authenticated user."""
    token = credentials.credentials
    return decode_token(token)

async def require_admin(current_user: TokenData = Depends(get_current_user)) -> TokenData:
    """Dependency to require admin privileges."""
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin privileges required"
        )
    return current_user

7. Pagination Utils (app/utils/pagination.py)

from typing import Generic, TypeVar, List
from pydantic import BaseModel
from fastapi import Query
from app.config import get_settings

settings = get_settings()
T = TypeVar('T')

class PaginationParams(BaseModel):
    """Pagination query parameters."""
    limit: int = Query(
        default=settings.DEFAULT_PAGE_SIZE,
        ge=1,
        le=settings.MAX_PAGE_SIZE,
        description="Number of items per page"
    )
    offset: int = Query(
        default=0,
        ge=0,
        description="Number of items to skip"
    )

class PaginatedResponse(BaseModel, Generic[T]):
    """Generic paginated response."""
    data: List[T]
    total: int
    limit: int
    offset: int
    has_more: bool

def paginate(
    items: List[T],
    total: int,
    params: PaginationParams
) -> PaginatedResponse[T]:
    """Create paginated response."""
    return PaginatedResponse(
        data=items,
        total=total,
        limit=params.limit,
        offset=params.offset,
        has_more=(params.offset + params.limit) < total
    )

8. User Service (app/services/user_service.py)

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from typing import Optional, List
from uuid import UUID
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
from app.utils.security import get_password_hash
from fastapi import HTTPException, status

class UserService:
    """Business logic for user operations."""

    @staticmethod
    async def create_user(db: AsyncSession, user_data: UserCreate) -> User:
        """Create new user."""
        # Check if email exists
        result = await db.execute(select(User).where(User.email == user_data.email))
        if result.scalar_one_or_none():
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="Email already registered"
            )

        # Create user
        user = User(
            email=user_data.email,
            full_name=user_data.full_name,
            hashed_password=get_password_hash(user_data.password)
        )
        db.add(user)
        await db.commit()
        await db.refresh(user)
        return user

    @staticmethod
    async def get_user_by_id(db: AsyncSession, user_id: UUID) -> Optional[User]:
        """Get user by ID."""
        result = await db.execute(select(User).where(User.id == user_id))
        return result.scalar_one_or_none()

    @staticmethod
    async def get_user_by_email(db: AsyncSession, email: str) -> Optional[User]:
        """Get user by email."""
        result = await db.execute(select(User).where(User.email == email))
        return result.scalar_one_or_none()

    @staticmethod
    async def list_users(
        db: AsyncSession,
        limit: int = 20,
        offset: int = 0
    ) -> tuple[List[User], int]:
        """List users with pagination."""
        # Get total count
        count_result = await db.execute(select(func.count(User.id)))
        total = count_result.scalar_one()

        # Get users
        result = await db.execute(
            select(User)
            .order_by(User.created_at.desc())
            .limit(limit)
            .offset(offset)
        )
        users = result.scalars().all()

        return list(users), total

    @staticmethod
    async def update_user(
        db: AsyncSession,
        user_id: UUID,
        user_data: UserUpdate
    ) -> Optional[User]:
        """Update user."""
        user = await UserService.get_user_by_id(db, user_id)
        if not user:
            return None

        update_data = user_data.model_dump(exclude_unset=True)
        for field, value in update_data.items():
            setattr(user, field, value)

        await db.commit()
        await db.refresh(user)
        return user

    @staticmethod
    async def delete_user(db: AsyncSession, user_id: UUID) -> bool:
        """Delete user."""
        user = await UserService.get_user_by_id(db, user_id)
        if not user:
            return False

        await db.delete(user)
        await db.commit()
        return True

9. Auth Router (app/routers/auth.py)

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.user import Token
from app.services.user_service import UserService
from app.utils.security import (
    verify_password,
    create_access_token,
    create_refresh_token
)
from app.config import get_settings

settings = get_settings()
router = APIRouter(prefix="/auth", tags=["Authentication"])

@router.post("/login", response_model=Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db)
):
    """
    Authenticate user and return JWT tokens.

    - **username**: User email address
    - **password**: User password
    """
    # Get user by email
    user = await UserService.get_user_by_email(db, form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )

    # Create tokens
    token_data = {
        "sub": str(user.id),
        "email": user.email,
        "is_admin": user.is_admin
    }

    access_token = create_access_token(token_data)
    refresh_token = create_refresh_token(token_data)

    return Token(
        access_token=access_token,
        refresh_token=refresh_token,
        expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60
    )

10. Users Router (app/routers/users.py)

from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.ext.asyncio import AsyncSession
from uuid import UUID
from typing import List
from app.database import get_db
from app.schemas.user import (
    UserCreate,
    UserUpdate,
    UserResponse,
    TokenData
)
from app.services.user_service import UserService
from app.utils.security import get_current_user, require_admin
from app.utils.pagination import (
    PaginationParams,
    PaginatedResponse,
    paginate
)

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

@router.post(
    "",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED
)
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db),
    _: TokenData = Depends(require_admin)  # Admin only
):
    """Create a new user (admin only)."""
    user = await UserService.create_user(db, user_data)
    return UserResponse.model_validate(user)

@router.get(
    "",
    response_model=PaginatedResponse[UserResponse]
)
async def list_users(
    pagination: PaginationParams = Depends(),
    db: AsyncSession = Depends(get_db),
    _: TokenData = Depends(get_current_user)
):
    """List all users with pagination."""
    users, total = await UserService.list_users(
        db,
        limit=pagination.limit,
        offset=pagination.offset
    )

    user_responses = [UserResponse.model_validate(u) for u in users]
    return paginate(user_responses, total, pagination)

@router.get(
    "/me",
    response_model=UserResponse
)
async def get_current_user_profile(
    current_user: TokenData = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    """Get current authenticated user's profile."""
    user = await UserService.get_user_by_id(db, current_user.user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return UserResponse.model_validate(user)

@router.get(
    "/{user_id}",
    response_model=UserResponse
)
async def get_user(
    user_id: UUID,
    db: AsyncSession = Depends(get_db),
    _: TokenData = Depends(get_current_user)
):
    """Get user by ID."""
    user = await UserService.get_user_by_id(db, user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return UserResponse.model_validate(user)

@router.patch(
    "/{user_id}",
    response_model=UserResponse
)
async def update_user(
    user_id: UUID,
    user_data: UserUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: TokenData = Depends(get_current_user)
):
    """Update user (own profile or admin)."""
    # Check permissions
    if user_id != current_user.user_id and not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Not authorized to update this user"
        )

    user = await UserService.update_user(db, user_id, user_data)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return UserResponse.model_validate(user)

@router.delete(
    "/{user_id}",
    status_code=status.HTTP_204_NO_CONTENT
)
async def delete_user(
    user_id: UUID,
    db: AsyncSession = Depends(get_db),
    _: TokenData = Depends(require_admin)  # Admin only
):
    """Delete user (admin only)."""
    deleted = await UserService.delete_user(db, user_id)
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )

11. Main Application (app/main.py)

from fastapi import FastAPI, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from contextlib import asynccontextmanager
import time
from app.config import get_settings
from app.routers import auth, users

settings = get_settings()

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan events."""
    # Startup
    print("Starting up...")
    yield
    # Shutdown
    print("Shutting down...")

app = FastAPI(
    title=settings.APP_NAME,
    version=settings.APP_VERSION,
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc"
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Request timing middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """Add X-Process-Time header to responses."""
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

# Global exception handlers
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Handle validation errors with RFC 9457 (Problem Details) format."""
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })

    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "type": "https://api.example.com/errors/validation",
            "title": "Validation Error",
            "status": 422,
            "detail": "Request validation failed",
            "errors": errors
        }
    )

# Health check
@app.get("/health", tags=["Health"])
async def health_check():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "version": settings.APP_VERSION
    }

# Include routers
app.include_router(auth.router, prefix="/api/v1")
app.include_router(users.router, prefix="/api/v1")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",
        port=8000,
        reload=settings.DEBUG
    )

12. Environment Variables (.env.example)

# App
APP_NAME="FastAPI App"
APP_VERSION="1.0.0"
DEBUG=False

# Database
DATABASE_URL="postgresql://user:password@localhost:5432/dbname"
ASYNC_DATABASE_URL="postgresql+asyncpg://user:password@localhost:5432/dbname"

# Security (CHANGE IN PRODUCTION)
SECRET_KEY="your-secret-key-here-change-in-production"
ALGORITHM="HS256"
ACCESS_TOKEN_EXPIRE_MINUTES=30
REFRESH_TOKEN_EXPIRE_DAYS=7

# CORS
CORS_ORIGINS=["http://localhost:3000","http://localhost:8080"]

# Rate Limiting
RATE_LIMIT_PER_MINUTE=60

# Pagination
DEFAULT_PAGE_SIZE=20
MAX_PAGE_SIZE=100

13. Database Migrations (Alembic)

# Initialize Alembic
alembic init alembic

# Edit alembic/env.py to import your models
# Then create first migration
alembic revision --autogenerate -m "Initial migration"

# Apply migrations
alembic upgrade head

14. Testing (tests/test_users.py)

import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_create_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/users",
            json={
                "email": "test@example.com",
                "full_name": "Test User",
                "password": "SecurePass123"
            }
        )
    assert response.status_code == 201
    assert response.json()["email"] == "test@example.com"

@pytest.mark.asyncio
async def test_login():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/auth/login",
            data={
                "username": "test@example.com",
                "password": "SecurePass123"
            }
        )
    assert response.status_code == 200
    assert "access_token" in response.json()

Running the Application

# Install dependencies
pip install -r requirements.txt

# Set up database
createdb your_database_name

# Run migrations
alembic upgrade head

# Start development server
uvicorn app.main:app --reload

# API docs available at:
# - Swagger UI: http://localhost:8000/docs
# - ReDoc: http://localhost:8000/redoc

Key Features

[check] Async/await: Full async support with AsyncSession [check] Type safety: Pydantic v2 for request/response validation [check] Authentication: JWT with access & refresh tokens [check] Authorization: Role-based access control (RBAC) [check] Database: SQLAlchemy 2.0 with Alembic migrations [check] Pagination: Cursor & offset pagination helpers [check] Error handling: RFC 9457 (Problem Details) format [check] OpenAPI: Auto-generated Swagger UI and ReDoc [check] Testing: pytest-asyncio setup [check] Security: Password hashing, CORS, rate limiting [check] Dependency injection: FastAPI's DI system [check] Configuration: Environment-based settings with Pydantic

Best Practices Applied

• Service layer pattern for business logic
• Repository pattern via SQLAlchemy
• Dependency injection for testability
• Async database operations
• Proper error handling and status codes
• Input validation with Pydantic
• JWT token security
• Database connection pooling
• Structured logging ready
• Type hints throughout