Implementing Api Patterns

API Patterns Skill

Purpose

Design and implement APIs using the optimal pattern and framework for the use case. Choose between REST, GraphQL, gRPC, and tRPC based on API consumers, performance requirements, and type safety needs.

When to Use This Skill

Use when:

• Building backend APIs for web, mobile, or service consumers
• Connecting frontend components (forms, tables, dashboards) to databases
• Implementing pagination, rate limiting, or caching strategies
• Generating OpenAPI documentation automatically
• Choosing between REST, GraphQL, gRPC, or tRPC patterns
• Integrating authentication and authorization
• Optimizing API performance and scalability

Quick Decision Framework

WHO CONSUMES YOUR API?
├─ PUBLIC/THIRD-PARTY DEVELOPERS → REST with OpenAPI
│  ├─ Python → FastAPI (auto-docs, 40k req/s)
│  ├─ TypeScript → Hono (edge-first, 50k req/s, 14KB)
│  ├─ Rust → Axum (140k req/s, <1ms latency)
│  └─ Go → Gin (100k+ req/s, mature ecosystem)
│
├─ FRONTEND TEAM (same org)
│  ├─ TypeScript full-stack? → tRPC (E2E type safety)
│  └─ Complex data needs? → GraphQL
│      ├─ Python → Strawberry
│      ├─ Rust → async-graphql
│      ├─ Go → gqlgen
│      └─ TypeScript → Pothos
│
├─ SERVICE-TO-SERVICE (microservices)
│  └─ High performance → gRPC
│      ├─ Rust → Tonic
│      ├─ Go → Connect-Go (browser-friendly)
│      └─ Python → grpcio
│
└─ MOBILE APPS
   ├─ Bandwidth constrained → GraphQL (request only needed fields)
   └─ Simple CRUD → REST (standard, well-understood)

REST Framework Selection

Python: FastAPI (Recommended)

Key Features: Auto OpenAPI docs, Pydantic v2 validation, async/await, 40k req/s

Basic Example:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items")
async def create_item(item: Item):
    return {"id": 1, **item.dict()}

See references/rest-design-principles.md for FastAPI patterns and examples/python-fastapi/.

TypeScript: Hono (Edge-First)

Key Features: 14KB bundle, runs on any runtime (Node/Deno/Bun/edge), Zod validation, 50k req/s

Basic Example:

import { Hono } from 'hono'
import { zValidator } from '@hono/zod-validator'
import { z } from 'zod'

const app = new Hono()
app.post('/items', zValidator('json', z.object({
  name: z.string(), price: z.number()
})), (c) => c.json({ id: 1, ...c.req.valid('json') }))

See references/rest-design-principles.md for Hono patterns and examples/typescript-hono/.

TypeScript: tRPC (Full-Stack Type Safety)

Key Features: Zero codegen, E2E type safety, React Query integration, WebSocket subscriptions

Basic Example:

import { initTRPC } from '@trpc/server'
import { z } from 'zod'

const t = initTRPC.create()
export const appRouter = t.router({
  createItem: t.procedure
    .input(z.object({ name: z.string(), price: z.number() }))
    .mutation(({ input }) => ({ id: '1', ...input }))
})
export type AppRouter = typeof appRouter

See references/trpc-setup-guide.md for setup patterns and examples/typescript-trpc/.

Rust: Axum (High Performance)

Key Features: Tower middleware, type-safe extractors, 140k req/s, compile-time verification

Basic Example:

use axum::{routing::post, Json, Router};
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct CreateItem { name: String, price: f64 }

#[derive(Serialize)]
struct Item { id: u64, name: String, price: f64 }

async fn create_item(Json(payload): Json<CreateItem>) -> Json<Item> {
    Json(Item { id: 1, name: payload.name, price: payload.price })
}

See references/rest-design-principles.md for Axum patterns and examples/rust-axum/.

Go: Gin (Mature Ecosystem)

Key Features: Largest Go ecosystem, 100k+ req/s, struct tag validation

Basic Example:

type Item struct {
    Name  string  `json:"name" binding:"required"`
    Price float64 `json:"price" binding:"required,gt=0"`
}

r := gin.Default()
r.POST("/items", func(c *gin.Context) {
    var item Item
    if c.ShouldBindJSON(&item); err != nil {
        c.JSON(400, gin.H{"error": err.Error()}); return
    }
    c.JSON(201, item)
})

See references/rest-design-principles.md for Gin patterns and examples/go-gin/.

Performance Benchmarks

Notes:

• Benchmarks assume single-core, JSON responses
• Actual performance varies with workload complexity
• Cold start only applies to serverless/edge deployments

Pagination Strategies

Cursor-Based (Recommended)

Advantages: Handles real-time changes, no skipped/duplicate records, scales to billions

FastAPI Example:

@app.get("/items")
async def list_items(cursor: Optional[str] = None, limit: int = 20):
    query = db.query(Item).filter(Item.id > cursor) if cursor else db.query(Item)
    items = query.limit(limit).all()
    return {
        "items": items,
        "next_cursor": items[-1].id if items else None,
        "has_more": len(items) == limit
    }

Offset-Based (Simple Cases Only)

Use only for static datasets (<10k records) with direct page access needs.

See references/pagination-patterns.md for complete patterns and frontend integration.

OpenAPI Documentation

FastAPI Example (Zero Config):

app = FastAPI(title="My API", version="1.0.0")

@app.post("/items", tags=["items"])
async def create_item(item: Item) -> Item:
    """Create item with name and price"""
    return item
# Docs at /docs, /redoc, /openapi.json

See references/openapi-documentation.md for framework-specific setup. Use scripts/generate_openapi.py to extract specs programmatically.

Frontend Integration Patterns

Forms → REST POST/PUT

Backend:

class UserCreate(BaseModel):
    email: EmailStr; name: str; age: int

@app.post("/api/users", status_code=201)
async def create_user(user: UserCreate):
    return {"id": 1, **user.dict()}

Frontend:

const res = await fetch('/api/users', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify(data)
})
if (!res.ok) throw new Error((await res.json()).detail)

Tables → GET with Pagination

See cursor pagination example above and references/pagination-patterns.md.

AI Chat → SSE Streaming

Backend:

from sse_starlette.sse import EventSourceResponse

@app.post("/api/chat")
async def chat(message: str):
    async def gen():
        for chunk in llm_stream(message):
            yield {"event": "message", "data": chunk}
    return EventSourceResponse(gen())

Frontend:

const es = new EventSource('/api/chat')
es.addEventListener('message', (e) => appendToChat(e.data))

See examples/ for complete integration examples with each frontend skill.

Rate Limiting

FastAPI Example (Token Bucket):

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/items")
@limiter.limit("100/minute")
async def list_items():
    return {"items": []}

See references/rate-limiting-strategies.md for sliding window, distributed patterns, and Redis implementation.

GraphQL Libraries

Use when frontend needs flexible data fetching or mobile apps have bandwidth constraints.

By Language:

• Python: Strawberry 0.287 (type-hint-based, async)
• Rust: async-graphql (high performance, tokio)
• Go: gqlgen (code generation from schema)
• TypeScript: Pothos (type-safe builder, no codegen)

See references/graphql-schema-design.md for schema patterns and N+1 prevention. See examples/graphql-strawberry/ for complete Python example.

gRPC for Microservices

Use for service-to-service communication with strong typing and high performance.

By Language:

• Rust: Tonic (async, type-safe, code generation)
• Go: Connect-Go (gRPC-compatible + browser-friendly)
• Python: grpcio (official implementation)
• TypeScript: @connectrpc/connect (browser + Node.js)

See references/grpc-protobuf-guide.md for Protocol Buffers guide. See examples/grpc-tonic/ for complete Rust example.

Additional Resources

References

• references/rest-design-principles.md - REST resource modeling, HTTP methods, status codes
• references/graphql-schema-design.md - Schema patterns, resolver optimization, N+1 prevention
• references/grpc-protobuf-guide.md - Proto3 syntax, service definitions, streaming
• references/trpc-setup-guide.md - Router patterns, middleware, Zod validation
• references/pagination-patterns.md - Cursor vs offset with mathematical explanation
• references/rate-limiting-strategies.md - Token bucket, sliding window, Redis
• references/caching-patterns.md - HTTP caching, application caching strategies
• references/versioning-strategies.md - URI, header, media type versioning
• references/openapi-documentation.md - Swagger/OpenAPI best practices by framework

Scripts (Token-Free Execution)

• scripts/generate_openapi.py - Generate OpenAPI spec from code
• scripts/validate_api_spec.py - Validate OpenAPI 3.1 compliance
• scripts/benchmark_endpoints.py - Load test API endpoints

Examples

• examples/python-fastapi/ - Complete FastAPI REST API
• examples/typescript-hono/ - Hono edge-first API
• examples/typescript-trpc/ - tRPC E2E type-safe API
• examples/rust-axum/ - Axum REST API
• examples/go-gin/ - Gin REST API
• examples/graphql-strawberry/ - Python GraphQL
• examples/grpc-tonic/ - Rust gRPC

Quick Reference

Choose REST when: Public API, standard CRUD, need caching, OpenAPI docs required Choose GraphQL when: Frontend needs flexible queries, mobile bandwidth constraints, complex nested data Choose gRPC when: Service-to-service communication, high performance, bidirectional streaming Choose tRPC when: TypeScript full-stack, same team owns frontend + backend, E2E type safety

Pagination: Always use cursor-based for production scale, offset-based only for simple cases Documentation: Prefer frameworks with automatic OpenAPI generation (FastAPI, Hono) Performance: Rust (Axum) for max throughput, Go (Gin) for maturity, Python (FastAPI) for DX

Go + Gin REST API Example

Production-ready REST API using Go and Gin framework with PostgreSQL.

Features

• Gin web framework (100k+ req/s)
• GORM or sqlc for database
• JWT authentication
• Middleware (CORS, logging, rate limiting)
• Struct validation
• OpenAPI/Swagger docs

Files

go-gin/
├── main.go
├── routes/
│   ├── auth.go
│   ├── users.go
│   └── posts.go
├── models/
│   └── user.go
├── middleware/
│   ├── auth.go
│   └── cors.go
├── handlers/
│   └── user_handler.go
├── go.mod
└── README.md

Quick Start

# Initialize module
go mod init github.com/yourusername/gin-api

# Install dependencies
go get -u github.com/gin-gonic/gin
go get -u gorm.io/gorm
go get -u gorm.io/driver/postgres

# Run
go run main.go

Example Code

package main

import (
    "github.com/gin-gonic/gin"
    "net/http"
)

type User struct {
    ID    uint   `json:"id"`
    Email string `json:"email" binding:"required,email"`
    Name  string `json:"name" binding:"required"`
}

func main() {
    r := gin.Default()

    // CORS middleware
    r.Use(func(c *gin.Context) {
        c.Writer.Header().Set("Access-Control-Allow-Origin", "*")
        c.Next()
    })

    // Routes
    r.GET("/users", getUsers)
    r.POST("/users", createUser)
    r.GET("/users/:id", getUser)

    r.Run(":8080")
}

func getUsers(c *gin.Context) {
    users := []User{{ID: 1, Email: "test@example.com", Name: "Test"}}
    c.JSON(http.StatusOK, users)
}

func createUser(c *gin.Context) {
    var user User
    if err := c.ShouldBindJSON(&user); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }
    c.JSON(http.StatusCreated, user)
}

func getUser(c *gin.Context) {
    id := c.Param("id")
    user := User{ID: 1, Email: "test@example.com", Name: "Test"}
    c.JSON(http.StatusOK, user)
}

Performance

• 100,000+ requests/second
• 1-2ms latency
• Compiled binary (~10MB)

Python GraphQL with Strawberry

GraphQL API using Strawberry (type-hint based) with FastAPI integration.

Features

• Strawberry GraphQL (async)
• FastAPI integration
• Type-hint based schema
• DataLoaders (N+1 prevention)
• Subscriptions (WebSocket)
• GraphQL Playground

Files

graphql-strawberry/
├── main.py                  # FastAPI + GraphQL
├── schema.py                # GraphQL schema
├── types/
│   ├── user.py
│   ├── post.py
│   └── comment.py
├── resolvers/
│   ├── user_resolver.py
│   └── post_resolver.py
├── dataloaders.py           # DataLoader setup
└── requirements.txt

Quick Start

# Install
pip install 'strawberry-graphql[fastapi]' fastapi uvicorn

# Run
uvicorn main:app --reload

Access GraphiQL: http://localhost:8000/graphql

Example Schema

import strawberry
from typing import List

@strawberry.type
class User:
    id: int
    email: str
    name: str
    posts: List['Post']

@strawberry.type
class Post:
    id: int
    title: str
    content: str
    author: User

@strawberry.type
class Query:
    @strawberry.field
    async def user(self, id: int) -> User:
        return await fetch_user(id)

    @strawberry.field
    async def users(self) -> List[User]:
        return await fetch_all_users()

@strawberry.type
class Mutation:
    @strawberry.mutation
    async def create_post(self, title: str, content: str) -> Post:
        return await create_post_in_db(title, content)

schema = strawberry.Schema(query=Query, mutation=Mutation)

FastAPI Integration

from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

app = FastAPI()

graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

DataLoader (N+1 Prevention)

from strawberry.dataloader import DataLoader

async def load_users(keys: List[int]) -> List[User]:
    users = await db.query("SELECT * FROM users WHERE id = ANY($1)", keys)
    return [users_dict[k] for k in keys]

user_loader = DataLoader(load_fn=load_users)

Rust gRPC with Tonic

High-performance gRPC service using Tonic framework.

Features

• Tonic (async gRPC)
• Protocol Buffers (proto3)
• Bidirectional streaming
• Tower middleware
• Automatic code generation

Files

grpc-tonic/
├── proto/
│   └── api.proto            # Protocol Buffers schema
├── src/
│   ├── main.rs
│   ├── server.rs            # gRPC server
│   └── client.rs            # gRPC client
├── build.rs                 # Proto compilation
└── Cargo.toml

Quick Start

# Install dependencies
cargo build

# Run server
cargo run --bin server

# Run client (separate terminal)
cargo run --bin client

Protocol Buffers Schema

// proto/api.proto
syntax = "proto3";

package api.v1;

service UserService {
  rpc GetUser (GetUserRequest) returns (User);
  rpc ListUsers (ListUsersRequest) returns (stream User);
  rpc CreateUser (CreateUserRequest) returns (User);
}

message User {
  int64 id = 1;
  string email = 2;
  string name = 3;
}

message GetUserRequest {
  int64 id = 1;
}

message ListUsersRequest {
  int32 limit = 1;
}

message CreateUserRequest {
  string email = 1;
  string name = 2;
}

Server Implementation

use tonic::{transport::Server, Request, Response, Status};

pub struct UserServiceImpl;

#[tonic::async_trait]
impl UserService for UserServiceImpl {
    async fn get_user(
        &self,
        request: Request<GetUserRequest>,
    ) -> Result<Response<User>, Status> {
        let req = request.into_inner();

        let user = User {
            id: req.id,
            email: "user@example.com".to_string(),
            name: "John Doe".to_string(),
        };

        Ok(Response::new(user))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "0.0.0.0:50051".parse()?;
    let service = UserServiceImpl::default();

    Server::builder()
        .add_service(UserServiceServer::new(service))
        .serve(addr)
        .await?;

    Ok(())
}

Client Usage

let mut client = UserServiceClient::connect("http://localhost:50051").await?;

let request = tonic::Request::new(GetUserRequest { id: 1 });
let response = client.get_user(request).await?;

println!("User: {:?}", response.into_inner());

Performance

• 50,000+ RPC/second
• <1ms latency
• Bidirectional streaming support
• HTTP/2 multiplexing

FastAPI Complete REST API Example

Complete REST API demonstrating FastAPI best practices.

Features

• Automatic OpenAPI documentation
• Pydantic v2 validation
• Cursor-based pagination
• Rate limiting (slowapi)
• CORS configuration
• Error handling
• Type hints throughout

Installation

pip install -r requirements.txt

Run

python main.py

Or with uvicorn directly:

uvicorn main:app --reload

Documentation

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc
• OpenAPI JSON: http://localhost:8000/openapi.json

API Endpoints

Health Check

• GET /health - Check service status

Items

• GET /items?cursor={cursor}&limit={limit} - List items (cursor pagination)
• GET /items/{id} - Get item by ID
• POST /items - Create item
• PUT /items/{id} - Update item
• DELETE /items/{id} - Delete item

Example Requests

List items (first page)

curl http://localhost:8000/items?limit=10

List items (next page)

curl http://localhost:8000/items?cursor=5&limit=10

Create item

curl -X POST http://localhost:8000/items \
  -H "Content-Type: application/json" \
  -d '{
    "name": "New Widget",
    "description": "A brand new widget",
    "price": 39.99
  }'

Get item

curl http://localhost:8000/items/1

Update item

curl -X PUT http://localhost:8000/items/1 \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Updated Widget",
    "description": "An updated widget",
    "price": 49.99
  }'

Delete item

curl -X DELETE http://localhost:8000/items/1

Rate Limits

• GET /items: 100 requests/minute
• POST /items: 10 requests/minute

Production Considerations

1. Database: Replace in-memory dict with real database (PostgreSQL, MongoDB) 2. Authentication: Add JWT or OAuth2 authentication 3. Caching: Add Redis for caching expensive queries 4. Logging: Add structured logging 5. Monitoring: Add Prometheus metrics 6. HTTPS: Run behind reverse proxy (Nginx) with SSL 7. Environment variables: Use .env for configuration

Frontend Integration

See implementing-api-patterns/SKILL.md for frontend integration examples with React, forms skill, and tables skill.

Rust Axum REST API Example

High-performance REST API using Axum, SQLx, and PostgreSQL with compile-time guarantees.

Features

• Axum web framework (140k req/s)
• SQLx compile-time SQL validation
• Tower middleware (CORS, logging)
• Type-safe extractors
• Async/await with Tokio

Files

rust-axum/
├── src/
│   ├── main.rs              # Server entry point
│   ├── routes.rs            # Route definitions
│   ├── handlers/
│   │   ├── mod.rs
│   │   └── users.rs
│   ├── models.rs            # Data types
│   ├── db.rs                # Database pool
│   └── error.rs             # Error handling
├── migrations/              # SQLx migrations
├── Cargo.toml
└── .env.example

Quick Start

# Setup database
cp .env.example .env
sqlx database create
sqlx migrate run

# Run
cargo run --release

Example Code

use axum::{
    extract::{Path, State},
    routing::{get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;

#[derive(Serialize, sqlx::FromRow)]
struct User {
    id: i64,
    email: String,
    name: String,
}

#[derive(Deserialize)]
struct CreateUser {
    email: String,
    name: String,
}

async fn list_users(State(pool): State<PgPool>) -> Json<Vec<User>> {
    let users = sqlx::query_as::<_, User>("SELECT id, email, name FROM users")
        .fetch_all(&pool)
        .await
        .unwrap();

    Json(users)
}

async fn create_user(
    State(pool): State<PgPool>,
    Json(payload): Json<CreateUser>,
) -> Json<User> {
    let user = sqlx::query_as::<_, User>(
        "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id, email, name"
    )
    .bind(&payload.email)
    .bind(&payload.name)
    .fetch_one(&pool)
    .await
    .unwrap();

    Json(user)
}

#[tokio::main]
async fn main() {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL").unwrap())
        .await
        .unwrap();

    let app = Router::new()
        .route("/users", get(list_users).post(create_user))
        .with_state(pool);

    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

Performance

• 140,000+ requests/second
• <1ms p99 latency
• 2-5MB memory
• Zero runtime overhead

TypeScript tRPC Full-Stack Example

End-to-end type-safe API using tRPC with Next.js and Prisma.

Features

• tRPC v11 (E2E type safety)
• Next.js 14 (App Router)
• Prisma (database)
• Zod validation
• React Query integration
• Zero codegen required

Files

typescript-trpc/
├── src/
│   ├── server/
│   │   ├── trpc.ts          # tRPC instance
│   │   ├── routers/
│   │   │   ├── _app.ts      # Root router
│   │   │   ├── auth.ts
│   │   │   └── users.ts
│   │   └── context.ts       # Request context
│   ├── app/
│   │   ├── api/trpc/[trpc]/route.ts
│   │   ├── layout.tsx
│   │   └── page.tsx
│   ├── utils/
│   │   └── trpc.ts          # Client setup
│   └── components/
│       └── UserList.tsx
├── prisma/schema.prisma
└── package.json

Quick Start

# Install
npm install

# Setup database
npx prisma migrate dev

# Run dev server
npm run dev

Server Setup

// src/server/trpc.ts
import { initTRPC } from '@trpc/server';
import { z } from 'zod';

const t = initTRPC.create();

export const router = t.router;
export const publicProcedure = t.procedure;

// src/server/routers/users.ts
export const userRouter = router({
  list: publicProcedure.query(async () => {
    return await prisma.user.findMany();
  }),

  byId: publicProcedure
    .input(z.object({ id: z.number() }))
    .query(async ({ input }) => {
      return await prisma.user.findUnique({ where: { id: input.id } });
    }),

  create: publicProcedure
    .input(z.object({ email: z.string().email(), name: z.string() }))
    .mutation(async ({ input }) => {
      return await prisma.user.create({ data: input });
    }),
});

// src/server/routers/_app.ts
export const appRouter = router({
  user: userRouter,
  auth: authRouter,
});

export type AppRouter = typeof appRouter;

Client Usage

'use client';

import { trpc } from '@/utils/trpc';

export function UserList() {
  const { data: users, isLoading } = trpc.user.list.useQuery();
  const createUser = trpc.user.create.useMutation();

  if (isLoading) return <div>Loading...</div>;

  return (
    <div>
      {users?.map(user => <div key={user.id}>{user.name}</div>)}

      <button onClick={() => createUser.mutate({
        email: 'new@example.com',
        name: 'New User'
      })}>
        Add User
      </button>
    </div>
  );
}

Type Safety

// ✅ Fully type-safe - no manual type definitions needed
const user = await trpc.user.byId.query({ id: 1 });
//    ^? User (inferred from server)

// ❌ TypeScript error if wrong type
await trpc.user.byId.query({ id: "wrong" });  // Error: Expected number

Benefits

• Zero API boilerplate
• Catch errors at compile time
• Autocomplete for all endpoints
• Refactoring safety (rename propagates)
• React Query integration built-in

API Patterns Skill

Comprehensive guide for designing and implementing APIs across REST, GraphQL, gRPC, and tRPC patterns.

Overview

This skill provides framework selection guidance, pagination strategies, OpenAPI documentation, and frontend integration patterns for Python, TypeScript, Rust, and Go.

Status: Production-ready Version: 1.0.0 Date: December 2025

Quick Start

For Users

1. Choose API pattern based on consumers:

• Public API → REST (FastAPI, Hono, Axum, Gin)
• TypeScript full-stack → tRPC
• Flexible data fetching → GraphQL
• Service-to-service → gRPC

2. Reference SKILL.md for decision frameworks and quick examples

3. Use references/ for deep-dive guides:

• rest-design-principles.md - REST best practices
• pagination-patterns.md - Cursor vs offset pagination
• trpc-setup-guide.md - tRPC E2E type safety
• graphql-schema-design.md - GraphQL schemas, N+1 prevention
• grpc-protobuf-guide.md - gRPC and Protocol Buffers
• openapi-documentation.md - Auto-generated docs

4. Explore examples/ for complete working projects

5. Run scripts/ for token-free validation:

• generate_openapi.py - Extract OpenAPI specs
• validate_api_spec.py - Validate OpenAPI 3.1
• benchmark_endpoints.py - Load test APIs

For Developers

See init.md for master plan and architecture decisions.

Structure

implementing-api-patterns/
├── SKILL.md                        # Main skill file (<500 lines)
├── README.md                       # This file
├── init.md                         # Master plan
├── references/                     # Detailed guides (one level deep)
│   ├── rest-design-principles.md
│   ├── pagination-patterns.md
│   ├── trpc-setup-guide.md
│   ├── graphql-schema-design.md
│   ├── grpc-protobuf-guide.md
│   └── openapi-documentation.md
├── examples/                       # Working code examples
│   ├── python-fastapi/
│   │   ├── main.py
│   │   └── requirements.txt
│   └── typescript-hono/
│       ├── index.ts
│       └── package.json
└── scripts/                        # Token-free utilities
    ├── generate_openapi.py
    ├── validate_api_spec.py
    └── benchmark_endpoints.py

Key Features

Multi-Paradigm Coverage

• REST: FastAPI (Python), Hono (TypeScript), Axum (Rust), Gin (Go)
• GraphQL: Strawberry (Python), Pothos (TypeScript), async-graphql (Rust), gqlgen (Go)
• gRPC: grpcio (Python), Connect-Go (Go), Tonic (Rust)
• tRPC: TypeScript E2E type safety with zero codegen

Context7 Research

All framework recommendations validated with Context7:

• FastAPI: /websites/fastapi_tiangolo (Score: 79.8, Snippets: 29,015)
• Hono: /llmstxt/hono_dev_llms_txt (Score: 92.1, Snippets: 1,817)
• tRPC: /trpc/trpc (Score: 92.7, Snippets: 900)
• Axum: /websites/rs_axum_axum (Score: 77.5, Snippets: 7,260)

Performance Benchmarks

Pagination Strategies

Cursor-based (Recommended for Scale):

• Scales to billions of records
• No skipped/duplicate records
• Handles real-time data changes
• Working examples in Python, TypeScript

Offset-based (Simple Cases):

• Direct page number access
• Simple implementation
• Use for static, small datasets

OpenAPI Auto-Generation

• FastAPI: Zero configuration, automatic docs at /docs, /redoc
• Hono: Middleware plugin with Swagger UI
• Axum: utoipa crate with annotations
• Gin: swaggo/swag with comment annotations

Frontend Integration

Explicit patterns for connecting to all frontend skills:

• forms → POST/PUT/PATCH endpoints
• tables → GET + cursor pagination
• dashboards → GET + SSE for real-time
• ai-chat → POST + SSE streaming
• search-filter → GraphQL flexible queries
• media → POST multipart file uploads

Examples

FastAPI (Python)

cd examples/python-fastapi
pip install -r requirements.txt
python main.py
# Visit http://localhost:8000/docs

Features:

• Automatic OpenAPI documentation
• Cursor-based pagination
• Rate limiting (slowapi)
• Pydantic v2 validation

Hono (TypeScript)

cd examples/typescript-hono
bun install
bun run dev
# Visit http://localhost:3000/docs

Features:

• Edge-first (14KB, <5ms cold start)
• Zod validation
• OpenAPI middleware
• Runs on any runtime (Node, Deno, Bun, Cloudflare Workers)

Scripts Usage

Generate OpenAPI Spec

python scripts/generate_openapi.py examples/python-fastapi/main.py openapi.json

Extracts OpenAPI specification without running the server.

Validate OpenAPI Spec

python scripts/validate_api_spec.py openapi.json

Validates against OpenAPI 3.1 schema, reports errors and warnings.

Benchmark Endpoints

python scripts/benchmark_endpoints.py http://localhost:8000 --requests 1000 --concurrency 10

Load tests API endpoints and reports:

• Requests per second
• Latency (avg, min, max, p50, p95, p99)
• Success/failure rates

Decision Framework

Choose API Pattern

WHO CONSUMES YOUR API?

PUBLIC/THIRD-PARTY → REST
├─ Python → FastAPI
├─ TypeScript → Hono
├─ Rust → Axum
└─ Go → Gin

FRONTEND TEAM (same org)
├─ TypeScript full-stack → tRPC
└─ Complex data needs → GraphQL

SERVICE-TO-SERVICE → gRPC
└─ High performance needed

MOBILE APPS
├─ Bandwidth constrained → GraphQL
└─ Simple CRUD → REST

Choose Framework by Language

Python: FastAPI (modern, auto-docs) > Flask (lightweight) > Django REST

TypeScript: Hono (edge-first) > tRPC (full-stack TS) > Express (legacy)

Rust: Axum (ergonomics) > Actix-web (max perf) > Rocket (easy DX)

Go: Gin (standard) > net/http (stdlib) > Echo/Fiber (enterprise)

Best Practices

1. Use cursor pagination for production APIs (not offset) 2. Auto-generate OpenAPI docs (FastAPI, Hono) 3. Validate with Zod/Pydantic (type-safe validation) 4. Implement rate limiting (prevent abuse) 5. Use gRPC for microservices (high performance) 6. Use tRPC for TypeScript full-stack (E2E type safety) 7. Document all endpoints (descriptions, examples) 8. Handle errors properly (RFC 7807 Problem Details) 9. Enable CORS (for browser clients) 10. Version APIs (URI versioning: /api/v1/)

Integration with Other Skills

forms Skill

Use POST/PUT endpoints with Pydantic/Zod validation for form submissions.

tables Skill

Implement cursor pagination for scalable table data fetching.

dashboards Skill

Combine REST endpoints with SSE for real-time dashboard updates.

ai-chat Skill

Use SSE streaming for LLM response streaming to chat interfaces.

search-filter Skill

Implement GraphQL for flexible filtering and field selection.

databases-* Skills

Connect API endpoints to database operations (PostgreSQL, MongoDB, etc.).

Production Checklist

• [ ] Authentication implemented (JWT, OAuth2, API keys)
• [ ] Rate limiting configured
• [ ] CORS properly configured
• [ ] Error handling comprehensive
• [ ] OpenAPI docs generated and accessible
• [ ] Pagination implemented (cursor-based preferred)
• [ ] Validation schemas defined (Pydantic, Zod)
• [ ] Database connection pooling configured
• [ ] Logging structured and comprehensive
• [ ] Metrics/monitoring integrated (Prometheus)
• [ ] HTTPS enforced (reverse proxy with SSL)
• [ ] Environment variables for configuration

Resources

Official Documentation

• FastAPI: https://fastapi.tiangolo.com
• Hono: https://hono.dev
• tRPC: https://trpc.io
• Axum: https://docs.rs/axum

Context7 Sources

• FastAPI: /websites/fastapi_tiangolo, /fastapi/fastapi
• Hono: /llmstxt/hono_dev_llms_txt, /honojs/hono
• tRPC: /trpc/trpc, /websites/trpc_io
• Axum: /websites/rs_axum_axum, /tokio-rs/axum

Related Skills

• forms - Form validation and submission
• tables - Data table integration with pagination
• dashboards - Real-time dashboard APIs
• ai-chat - SSE streaming for chat interfaces
• databases-sql - SQL database integration
• auth-security - Authentication and authorization

Version History

License

MIT License - See repository root for details

---

Built with Claude Code following Anthropic's best practices for Skills development.

Caching Patterns for APIs

Caching reduces latency, improves scalability, and decreases database load by storing frequently accessed data.

Table of Contents

• Cache Layers
• HTTP Caching (Browser/CDN)
• Cache-Control Headers
• ETag (Conditional Requests)
• Last-Modified (Time-based Caching)
• Application Caching (Redis)
• Simple Key-Value Cache
• Cache-Aside Pattern (Recommended)
• Write-Through Cache
• Write-Behind Cache (Lazy Write)
• Cache Invalidation Strategies
• Time-based (TTL)
• Event-based Invalidation
• Cache Tags
• Cache Warming
• Cache Stampede Prevention
• Multi-Level Caching
• Cache Performance Monitoring
• GraphQL-Specific Caching
• Recommendation Matrix

Cache Layers

Client
  ↓ (HTTP Cache)
CDN/Edge Cache
  ↓ (Application Cache)
API Server
  ↓ (Database Query Cache)
Database

HTTP Caching (Browser/CDN)

Cache-Control Headers

Public, Long-lived (Static Assets):

from fastapi import Response

@app.get("/api/config")
async def get_config(response: Response):
    response.headers["Cache-Control"] = "public, max-age=3600, immutable"
    response.headers["ETag"] = '"v1.2.3"'
    return {"version": "1.2.3", "features": ["dark-mode"]}

Private, Short-lived (User Data):

@app.get("/api/user/profile")
async def get_profile(response: Response):
    response.headers["Cache-Control"] = "private, max-age=60"
    return {"name": "User", "email": "user@example.com"}

No Cache (Sensitive/Real-time):

@app.get("/api/user/balance")
async def get_balance(response: Response):
    response.headers["Cache-Control"] = "no-store, must-revalidate"
    response.headers["Pragma"] = "no-cache"
    return {"balance": 1234.56}

ETag (Conditional Requests)

Server-side:

from fastapi import Request, Response, HTTPException
import hashlib
import json

@app.get("/api/items/{item_id}")
async def get_item(item_id: int, request: Request, response: Response):
    item = await db.get_item(item_id)
    item_json = json.dumps(item, sort_keys=True)

    # Generate ETag from content
    etag = f'"{hashlib.md5(item_json.encode()).hexdigest()}"'

    # Check If-None-Match header
    if request.headers.get("If-None-Match") == etag:
        raise HTTPException(status_code=304)  # Not Modified

    response.headers["ETag"] = etag
    response.headers["Cache-Control"] = "private, max-age=60"
    return item

Client-side:

// First request
const res1 = await fetch('/api/items/1')
const etag = res1.headers.get('ETag')
const data = await res1.json()

// Subsequent request
const res2 = await fetch('/api/items/1', {
  headers: { 'If-None-Match': etag }
})

if (res2.status === 304) {
  // Use cached data
  console.log('Using cached data:', data)
} else {
  // Update cache
  const newData = await res2.json()
}

Last-Modified (Time-based Caching)

from datetime import datetime

@app.get("/api/items/{item_id}")
async def get_item(item_id: int, request: Request, response: Response):
    item = await db.get_item(item_id)

    last_modified = item.updated_at
    response.headers["Last-Modified"] = last_modified.strftime("%a, %d %b %Y %H:%M:%S GMT")

    # Check If-Modified-Since
    if_modified_since = request.headers.get("If-Modified-Since")
    if if_modified_since:
        client_date = datetime.strptime(if_modified_since, "%a, %d %b %Y %H:%M:%S GMT")
        if client_date >= last_modified:
            raise HTTPException(status_code=304)

    return item

Application Caching (Redis)

Simple Key-Value Cache

FastAPI + Redis:

from redis import Redis
import json

redis_client = Redis(host='localhost', port=6379, decode_responses=True)

@app.get("/api/items/{item_id}")
async def get_item(item_id: int):
    cache_key = f"item:{item_id}"

    # Try cache first
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # Cache miss - query database
    item = await db.get_item(item_id)

    # Store in cache (TTL: 5 minutes)
    redis_client.setex(cache_key, 300, json.dumps(item))

    return item

Cache-Aside Pattern (Recommended)

Decorator Approach:

from functools import wraps
import pickle

def cache_aside(ttl: int = 300, key_prefix: str = ""):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Generate cache key from function name and arguments
            cache_key = f"{key_prefix}:{func.__name__}:{hash((args, tuple(kwargs.items())))}"

            # Try cache
            cached = redis_client.get(cache_key)
            if cached:
                return pickle.loads(cached)

            # Execute function
            result = await func(*args, **kwargs)

            # Store in cache
            redis_client.setex(cache_key, ttl, pickle.dumps(result))

            return result
        return wrapper
    return decorator

@app.get("/api/items")
@cache_aside(ttl=300, key_prefix="items_list")
async def list_items(limit: int = 20, offset: int = 0):
    return await db.query(Item).limit(limit).offset(offset).all()

Write-Through Cache

Update cache on write:

@app.post("/api/items")
async def create_item(item: ItemCreate):
    # Write to database
    new_item = await db.create(item)

    # Immediately update cache
    cache_key = f"item:{new_item.id}"
    redis_client.setex(cache_key, 300, json.dumps(new_item.to_dict()))

    # Invalidate list cache
    redis_client.delete("items_list:*")

    return new_item

Write-Behind Cache (Lazy Write)

Queue writes for background processing:

from asyncio import Queue

write_queue = Queue()

@app.post("/api/items")
async def create_item(item: ItemCreate):
    # Update cache immediately
    item_id = generate_id()
    cache_key = f"item:{item_id}"
    redis_client.setex(cache_key, 300, json.dumps(item.dict()))

    # Queue database write
    await write_queue.put(("create", item))

    return {"id": item_id, **item.dict()}

# Background worker
async def write_worker():
    while True:
        operation, data = await write_queue.get()
        try:
            if operation == "create":
                await db.create(data)
        except Exception as e:
            logger.error(f"Write failed: {e}")

Cache Invalidation Strategies

Time-based (TTL)

Fixed expiration:

redis_client.setex("key", 300, "value")  # Expires in 5 minutes

Sliding expiration:

redis_client.set("key", "value")
redis_client.expire("key", 300)  # Refresh TTL on access

Event-based Invalidation

Invalidate on write:

@app.put("/api/items/{item_id}")
async def update_item(item_id: int, item: ItemUpdate):
    # Update database
    updated_item = await db.update(item_id, item)

    # Invalidate specific item cache
    redis_client.delete(f"item:{item_id}")

    # Invalidate related caches
    redis_client.delete(f"user:{updated_item.user_id}:items")

    return updated_item

Cache Tags

Group related cache entries:

from typing import List

class TaggedCache:
    def __init__(self, redis_client):
        self.redis = redis_client

    def set_with_tags(self, key: str, value: str, tags: List[str], ttl: int = 300):
        # Store value
        self.redis.setex(key, ttl, value)

        # Add to tag sets
        for tag in tags:
            self.redis.sadd(f"tag:{tag}", key)
            self.redis.expire(f"tag:{tag}", ttl)

    def invalidate_by_tag(self, tag: str):
        # Get all keys with this tag
        keys = self.redis.smembers(f"tag:{tag}")

        # Delete all tagged keys
        if keys:
            self.redis.delete(*keys)

        # Delete tag set
        self.redis.delete(f"tag:{tag}")

cache = TaggedCache(redis_client)

@app.get("/api/items/{item_id}")
async def get_item(item_id: int):
    cache_key = f"item:{item_id}"
    cached = cache.redis.get(cache_key)
    if cached:
        return json.loads(cached)

    item = await db.get_item(item_id)
    cache.set_with_tags(
        cache_key,
        json.dumps(item),
        tags=["items", f"user:{item.user_id}"],
        ttl=300
    )
    return item

@app.delete("/api/users/{user_id}")
async def delete_user(user_id: int):
    await db.delete_user(user_id)
    # Invalidate all items for this user
    cache.invalidate_by_tag(f"user:{user_id}")

Cache Warming

Preload cache on startup:

@app.on_event("startup")
async def warm_cache():
    # Load frequently accessed items
    popular_items = await db.query(Item).order_by(Item.views.desc()).limit(100).all()

    for item in popular_items:
        cache_key = f"item:{item.id}"
        redis_client.setex(cache_key, 3600, json.dumps(item.to_dict()))

    logger.info(f"Warmed cache with {len(popular_items)} items")

Cache Stampede Prevention

Problem: Many requests hit database simultaneously when cache expires.

Solution: Lock-based Approach:

import asyncio

async def get_with_lock(key: str, fetch_func, ttl: int = 300):
    # Try cache
    cached = redis_client.get(key)
    if cached:
        return json.loads(cached)

    # Acquire lock
    lock_key = f"lock:{key}"
    lock_acquired = redis_client.set(lock_key, "1", nx=True, ex=10)

    if lock_acquired:
        # This request fetches fresh data
        try:
            data = await fetch_func()
            redis_client.setex(key, ttl, json.dumps(data))
            return data
        finally:
            redis_client.delete(lock_key)
    else:
        # Wait for other request to populate cache
        for _ in range(10):
            await asyncio.sleep(0.1)
            cached = redis_client.get(key)
            if cached:
                return json.loads(cached)

        # Fallback: fetch anyway
        return await fetch_func()

@app.get("/api/expensive-query")
async def expensive_query():
    return await get_with_lock(
        "expensive_query_result",
        lambda: db.complex_aggregation(),
        ttl=600
    )

Solution: Probabilistic Early Expiration:

import random
import time

async def probabilistic_cache(key: str, fetch_func, ttl: int = 300):
    cached_data = redis_client.get(key)
    if not cached_data:
        # Cache miss
        data = await fetch_func()
        redis_client.setex(key, ttl, json.dumps({"data": data, "stored_at": time.time()}))
        return data

    cache_entry = json.loads(cached_data)
    stored_at = cache_entry["stored_at"]
    age = time.time() - stored_at

    # Probability of early refresh increases with age
    refresh_probability = age / ttl

    if random.random() < refresh_probability:
        # Refresh cache early
        data = await fetch_func()
        redis_client.setex(key, ttl, json.dumps({"data": data, "stored_at": time.time()}))
        return data

    return cache_entry["data"]

Multi-Level Caching

In-memory + Redis:

from cachetools import TTLCache
import asyncio

# L1: In-memory cache (fast, small)
l1_cache = TTLCache(maxsize=1000, ttl=60)

# L2: Redis cache (slower, larger)
redis_client = Redis()

async def multi_level_get(key: str, fetch_func):
    # Check L1 (in-memory)
    if key in l1_cache:
        return l1_cache[key]

    # Check L2 (Redis)
    cached = redis_client.get(key)
    if cached:
        data = json.loads(cached)
        l1_cache[key] = data  # Populate L1
        return data

    # Cache miss - fetch from source
    data = await fetch_func()

    # Populate both caches
    l1_cache[key] = data
    redis_client.setex(key, 300, json.dumps(data))

    return data

Cache Performance Monitoring

Track cache hit rate:

from prometheus_client import Counter, Histogram

cache_hits = Counter('cache_hits_total', 'Number of cache hits')
cache_misses = Counter('cache_misses_total', 'Number of cache misses')
cache_latency = Histogram('cache_latency_seconds', 'Cache operation latency')

async def get_cached(key: str, fetch_func):
    with cache_latency.time():
        cached = redis_client.get(key)
        if cached:
            cache_hits.inc()
            return json.loads(cached)

        cache_misses.inc()
        data = await fetch_func()
        redis_client.setex(key, 300, json.dumps(data))
        return data

GraphQL-Specific Caching

DataLoader pattern (N+1 prevention):

from strawberry.dataloader import DataLoader

async def load_users(keys: List[int]) -> List[User]:
    # Batch database query
    users = await db.query(User).filter(User.id.in_(keys)).all()
    user_map = {user.id: user for user in users}

    # Return in same order as keys
    return [user_map.get(key) for key in keys]

user_loader = DataLoader(load_fn=load_users)

@strawberry.type
class Query:
    @strawberry.field
    async def user(self, id: int) -> User:
        return await user_loader.load(id)

Recommendation Matrix

General Guidelines:

• Static data: Long TTL (hours/days)
• User data: Medium TTL (minutes)
• Real-time data: Short TTL (seconds) or no cache
• Always measure: Track hit rate, latency, memory usage

GraphQL Schema Design

Overview

GraphQL provides a query language for APIs with strong typing and flexible data fetching. This guide covers schema patterns, resolver optimization, and N+1 query prevention.

Table of Contents

• Core Concepts
• Schema Definition
• Resolver Patterns
• N+1 Problem and DataLoader
• Error Handling
• Pagination
• Subscriptions

Core Concepts

When to Use GraphQL

✅ Use GraphQL when:

• Frontend needs flexible data fetching
• Mobile apps with bandwidth constraints
• Complex, nested data requirements
• Over-fetching/under-fetching problems exist
• Multiple clients need different data shapes

❌ Avoid GraphQL when:

• Simple CRUD operations suffice
• No need for flexible queries
• Team lacks GraphQL experience
• Caching requirements are complex

GraphQL vs REST

Schema Definition

Python: Strawberry

import strawberry
from typing import List, Optional

@strawberry.type
class User:
    id: strawberry.ID
    name: str
    email: str
    posts: List['Post']

@strawberry.type
class Post:
    id: strawberry.ID
    title: str
    content: str
    author: User

@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: strawberry.ID) -> Optional[User]:
        return get_user(id)

    @strawberry.field
    def users(self) -> List[User]:
        return get_all_users()

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, name: str, email: str) -> User:
        return create_user_in_db(name, email)

schema = strawberry.Schema(query=Query, mutation=Mutation)

TypeScript: Pothos

import SchemaBuilder from '@pothos/core'

const builder = new SchemaBuilder({})

builder.objectType('User', {
  fields: (t) => ({
    id: t.exposeID('id'),
    name: t.exposeString('name'),
    email: t.exposeString('email'),
    posts: t.field({
      type: [Post],
      resolve: (user) => getPostsByUserId(user.id)
    })
  })
})

builder.objectType('Post', {
  fields: (t) => ({
    id: t.exposeID('id'),
    title: t.exposeString('title'),
    content: t.exposeString('content'),
    author: t.field({
      type: User,
      resolve: (post) => getUserById(post.authorId)
    })
  })
})

builder.queryType({
  fields: (t) => ({
    user: t.field({
      type: User,
      nullable: true,
      args: { id: t.arg.id({ required: true }) },
      resolve: (parent, args) => getUserById(args.id)
    }),
    users: t.field({
      type: [User],
      resolve: () => getAllUsers()
    })
  })
})

builder.mutationType({
  fields: (t) => ({
    createUser: t.field({
      type: User,
      args: {
        name: t.arg.string({ required: true }),
        email: t.arg.string({ required: true })
      },
      resolve: (parent, args) => createUser(args)
    })
  })
})

const schema = builder.toSchema()

Rust: async-graphql

use async_graphql::{Object, Schema, EmptySubscription};

struct User {
    id: i32,
    name: String,
    email: String,
}

#[Object]
impl User {
    async fn id(&self) -> i32 { self.id }
    async fn name(&self) -> &str { &self.name }
    async fn email(&self) -> &str { &self.email }

    async fn posts(&self) -> Vec<Post> {
        get_posts_by_user_id(self.id)
    }
}

struct Query;

#[Object]
impl Query {
    async fn user(&self, id: i32) -> Option<User> {
        get_user(id)
    }

    async fn users(&self) -> Vec<User> {
        get_all_users()
    }
}

struct Mutation;

#[Object]
impl Mutation {
    async fn create_user(&self, name: String, email: String) -> User {
        create_user_in_db(name, email)
    }
}

let schema = Schema::build(Query, Mutation, EmptySubscription).finish();

Resolver Patterns

Field Resolvers

Resolve individual fields (not just top-level queries):

@strawberry.type
class User:
    id: strawberry.ID
    name: str
    email: str

    @strawberry.field
    async def posts(self) -> List['Post']:
        # Resolve posts field when requested
        return await db.posts.find(author_id=self.id)

    @strawberry.field
    async def followers(self) -> List['User']:
        # Resolve followers only if requested
        follower_ids = await db.follows.find(following_id=self.id)
        return await db.users.find(id__in=follower_ids)

Client can request exactly what they need:

query {
  user(id: "123") {
    name
    # posts field NOT requested - resolver NOT called
  }
}

query {
  user(id: "123") {
    name
    posts {  # NOW posts resolver is called
      title
    }
  }
}

Context Pattern

Share data across resolvers:

from strawberry.fastapi import GraphQLRouter
from fastapi import FastAPI, Depends

async def get_context(
    request: Request,
    db = Depends(get_db),
    user = Depends(get_current_user)
):
    return {
        "request": request,
        "db": db,
        "user": user
    }

@strawberry.type
class Query:
    @strawberry.field
    async def current_user(self, info: strawberry.Info) -> User:
        # Access context
        return info.context["user"]

app = FastAPI()
graphql_app = GraphQLRouter(schema, context_getter=get_context)
app.include_router(graphql_app, prefix="/graphql")

N+1 Problem and DataLoader

The N+1 Problem

Problem: Fetching a list and then fetching related data for each item results in N+1 queries.

# BAD: N+1 queries
@strawberry.type
class User:
    id: strawberry.ID
    name: str

    @strawberry.field
    async def posts(self) -> List[Post]:
        # If querying 10 users, this runs 10 times!
        return await db.posts.find(author_id=self.id)

# Query
query {
  users {        # 1 query
    name
    posts {      # N queries (one per user)
      title
    }
  }
}
# Total: 1 + N queries

Solution: DataLoader

DataLoader batches and caches requests:

from strawberry.dataloader import DataLoader

async def load_posts_batch(user_ids: List[int]) -> List[List[Post]]:
    # Single query for all users' posts
    posts = await db.posts.find(author_id__in=user_ids)

    # Group by user_id
    posts_by_user = {}
    for post in posts:
        if post.author_id not in posts_by_user:
            posts_by_user[post.author_id] = []
        posts_by_user[post.author_id].append(post)

    # Return in same order as user_ids
    return [posts_by_user.get(uid, []) for uid in user_ids]

# Create DataLoader
posts_loader = DataLoader(load_fn=load_posts_batch)

@strawberry.type
class User:
    id: strawberry.ID
    name: str

    @strawberry.field
    async def posts(self, info: strawberry.Info) -> List[Post]:
        # DataLoader batches requests
        loader = info.context["posts_loader"]
        return await loader.load(self.id)

# In context
async def get_context():
    return {
        "posts_loader": DataLoader(load_fn=load_posts_batch)
    }

# Now querying 10 users = 2 queries total (1 for users, 1 for all posts)

TypeScript DataLoader

import DataLoader from 'dataloader'

const postsLoader = new DataLoader(async (userIds: number[]) => {
  // Single query for all users
  const posts = await db.post.findMany({
    where: { authorId: { in: userIds } }
  })

  // Group by authorId
  const postsByUser = new Map<number, Post[]>()
  posts.forEach(post => {
    if (!postsByUser.has(post.authorId)) {
      postsByUser.set(post.authorId, [])
    }
    postsByUser.get(post.authorId)!.push(post)
  })

  // Return in order
  return userIds.map(id => postsByUser.get(id) || [])
})

// In resolver
builder.objectType('User', {
  fields: (t) => ({
    posts: t.field({
      type: [Post],
      resolve: async (user, args, ctx) => {
        return ctx.postsLoader.load(user.id)
      }
    })
  })
})

Error Handling

Field-Level Errors

GraphQL allows partial success - some fields can error while others succeed:

@strawberry.type
class Query:
    @strawberry.field
    async def user(self, id: strawberry.ID) -> Optional[User]:
        user = await db.users.find_one(id=id)
        if not user:
            # Return null - error in response
            return None
        return user

    @strawberry.field
    async def risky_data(self) -> str:
        # Field raises error but other fields still resolve
        raise Exception("This field failed")

Response:

{
  "data": {
    "user": { "name": "Alice" },
    "riskyData": null
  },
  "errors": [
    {
      "message": "This field failed",
      "path": ["riskyData"]
    }
  ]
}

Custom Errors

from strawberry.types import Info

class NotFoundError(Exception):
    pass

class PermissionError(Exception):
    pass

@strawberry.type
class Query:
    @strawberry.field
    async def user(self, id: strawberry.ID, info: Info) -> User:
        user = await db.users.find_one(id=id)

        if not user:
            raise NotFoundError(f"User {id} not found")

        if not can_view_user(info.context["user"], user):
            raise PermissionError("Cannot view this user")

        return user

Pagination

Relay Cursor Connections

Standard pattern for pagination:

@strawberry.type
class PageInfo:
    has_next_page: bool
    has_previous_page: bool
    start_cursor: Optional[str]
    end_cursor: Optional[str]

@strawberry.type
class UserEdge:
    cursor: str
    node: User

@strawberry.type
class UserConnection:
    edges: List[UserEdge]
    page_info: PageInfo

@strawberry.type
class Query:
    @strawberry.field
    async def users(
        self,
        first: Optional[int] = None,
        after: Optional[str] = None
    ) -> UserConnection:
        # Decode cursor
        start_id = decode_cursor(after) if after else 0

        # Fetch first + 1 to check for next page
        users = await db.users.find(
            id__gt=start_id
        ).limit((first or 20) + 1)

        has_next = len(users) > (first or 20)
        if has_next:
            users = users[:first or 20]

        edges = [
            UserEdge(
                cursor=encode_cursor(user.id),
                node=user
            )
            for user in users
        ]

        return UserConnection(
            edges=edges,
            page_info=PageInfo(
                has_next_page=has_next,
                has_previous_page=after is not None,
                start_cursor=edges[0].cursor if edges else None,
                end_cursor=edges[-1].cursor if edges else None
            )
        )

# Query
query {
  users(first: 10, after: "cursor123") {
    edges {
      cursor
      node {
        name
        email
      }
    }
    pageInfo {
      hasNextPage
      endCursor
    }
  }
}

Offset Pagination (Simpler Alternative)

@strawberry.type
class UserPage:
    items: List[User]
    total: int
    page: int
    per_page: int

@strawberry.type
class Query:
    @strawberry.field
    async def users(
        self,
        page: int = 1,
        per_page: int = 20
    ) -> UserPage:
        offset = (page - 1) * per_page
        users = await db.users.find().skip(offset).limit(per_page)
        total = await db.users.count()

        return UserPage(
            items=users,
            total=total,
            page=page,
            per_page=per_page
        )

Subscriptions

Real-time updates via WebSocket:

import strawberry
from typing import AsyncIterator

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def post_created(self) -> AsyncIterator[Post]:
        # Subscribe to events
        async for post in event_stream('post_created'):
            yield post

    @strawberry.subscription
    async def comment_added(self, post_id: strawberry.ID) -> AsyncIterator[Comment]:
        async for comment in event_stream('comment_added'):
            if comment.post_id == post_id:
                yield comment

schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    subscription=Subscription
)

Client subscription:

subscription {
  postCreated {
    id
    title
    author {
      name
    }
  }
}

Best Practices

1. Use DataLoader for N+1 prevention - Essential for performance 2. Design schema from client perspective - Think about what data clients need 3. Avoid exposing database structure - Schema should match domain model 4. Use pagination for lists - Relay connections for consistency 5. Field-level authorization - Check permissions in resolvers 6. Deprecate fields instead of removing - GraphQL introspection shows deprecated fields 7. Limit query complexity - Prevent malicious deep queries 8. Enable query depth limiting - Prevent overly nested queries 9. Cache at field level - Use DataLoader caching 10. Document schema - Use descriptions for fields and types

Query Complexity Analysis

Prevent expensive queries:

from strawberry.extensions import QueryDepthLimiter

schema = strawberry.Schema(
    query=Query,
    extensions=[
        QueryDepthLimiter(max_depth=10)
    ]
)

This prevents queries like:

query {
  user {
    posts {
      author {
        posts {
          author {
            posts {  # Too deep!
              # ...
            }
          }
        }
      }
    }
  }
}

gRPC and Protocol Buffers Guide

Overview

gRPC is a high-performance RPC framework using Protocol Buffers for service-to-service communication. This guide covers proto3 syntax, service definitions, streaming patterns, and implementation examples.

Table of Contents

• Core Concepts
• Protocol Buffers (Proto3)
• Service Definitions
• Streaming Patterns
• Rust: Tonic
• Go: Connect-Go
• Python: grpcio
• Error Handling

Core Concepts

When to Use gRPC

✅ Use gRPC when:

• Service-to-service communication (microservices)
• High-performance requirements
• Strong typing and contract enforcement needed
• Bidirectional streaming needed
• Polyglot services (multiple languages)

❌ Avoid gRPC when:

• Browser clients (limited support, use Connect-Go for browser compatibility)
• Simple CRUD APIs (REST simpler)
• Human-readable debugging required (binary protocol)
• Public APIs (REST more accessible)

gRPC vs REST

Protocol Buffers (Proto3)

Basic Syntax

// user.proto
syntax = "proto3";

package example.v1;

// Message definition
message User {
  string id = 1;
  string name = 2;
  string email = 3;
  int32 age = 4;
  Role role = 5;
  repeated string tags = 6;  // List of strings
  google.protobuf.Timestamp created_at = 7;
}

// Enum definition
enum Role {
  ROLE_UNSPECIFIED = 0;  // First value must be 0
  ROLE_USER = 1;
  ROLE_ADMIN = 2;
  ROLE_MODERATOR = 3;
}

Field Numbers

Rules:

• Must be unique within a message
• Cannot be changed once in use (breaks compatibility)
• 1-15 use 1 byte (use for frequently set fields)
• 16-2047 use 2 bytes
• 19000-19999 are reserved

Data Types

Nested Messages

message User {
  string id = 1;
  string name = 2;
  Address address = 3;

  message Address {
    string street = 1;
    string city = 2;
    string country = 3;
  }
}

Maps

message UserProfile {
  string user_id = 1;
  map<string, string> metadata = 2;  // Key-value pairs
}

Oneof (Union Types)

message SearchRequest {
  string query = 1;

  oneof filter {
    string tag = 2;
    string category = 3;
    int32 year = 4;
  }
}

Importing Other Protos

import "google/protobuf/timestamp.proto";
import "common/types.proto";

message Post {
  string id = 1;
  string title = 2;
  google.protobuf.Timestamp created_at = 3;
  common.User author = 4;
}

Service Definitions

Basic Service

syntax = "proto3";

package user.v1;

service UserService {
  // Unary RPC (single request, single response)
  rpc GetUser(GetUserRequest) returns (GetUserResponse);

  rpc CreateUser(CreateUserRequest) returns (CreateUserResponse);

  rpc UpdateUser(UpdateUserRequest) returns (UpdateUserResponse);

  rpc DeleteUser(DeleteUserRequest) returns (DeleteUserResponse);

  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
}

message GetUserRequest {
  string id = 1;
}

message GetUserResponse {
  User user = 1;
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
}

message CreateUserResponse {
  User user = 1;
}

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

message ListUsersResponse {
  repeated User users = 1;
  string next_page_token = 2;
}

message User {
  string id = 1;
  string name = 2;
  string email = 3;
  google.protobuf.Timestamp created_at = 4;
}

Streaming Patterns

Server Streaming

Server sends multiple responses to single client request:

service LogService {
  // Server streams log entries
  rpc StreamLogs(StreamLogsRequest) returns (stream LogEntry);
}

message StreamLogsRequest {
  string service_name = 1;
  google.protobuf.Timestamp start_time = 2;
}

message LogEntry {
  google.protobuf.Timestamp timestamp = 1;
  string level = 2;
  string message = 3;
}

Client Streaming

Client sends multiple requests, server responds once:

service UploadService {
  // Client streams file chunks
  rpc UploadFile(stream FileChunk) returns (UploadResponse);
}

message FileChunk {
  bytes data = 1;
}

message UploadResponse {
  string file_id = 1;
  int64 size = 2;
}

Bidirectional Streaming

Both client and server stream messages:

service ChatService {
  // Bidirectional chat
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message ChatMessage {
  string user_id = 1;
  string message = 2;
  google.protobuf.Timestamp timestamp = 3;
}

Rust: Tonic

Installation

# Cargo.toml
[dependencies]
tonic = "0.11"
prost = "0.12"
tokio = { version = "1", features = ["full"] }

[build-dependencies]
tonic-build = "0.11"

Build Script

// build.rs
fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_build::compile_protos("proto/user.proto")?;
    Ok(())
}

Server Implementation

use tonic::{transport::Server, Request, Response, Status};

pub mod user {
    tonic::include_proto!("user.v1");
}

use user::user_service_server::{UserService, UserServiceServer};
use user::{GetUserRequest, GetUserResponse, User};

#[derive(Default)]
pub struct MyUserService {}

#[tonic::async_trait]
impl UserService for MyUserService {
    async fn get_user(
        &self,
        request: Request<GetUserRequest>
    ) -> Result<Response<GetUserResponse>, Status> {
        let req = request.into_inner();

        let user = match get_user_from_db(&req.id).await {
            Some(u) => u,
            None => return Err(Status::not_found("User not found"))
        };

        Ok(Response::new(GetUserResponse {
            user: Some(user)
        }))
    }

    async fn list_users(
        &self,
        request: Request<ListUsersRequest>
    ) -> Result<Response<ListUsersResponse>, Status> {
        let req = request.into_inner();
        let users = get_users_from_db(req.page_size, &req.page_token).await;

        Ok(Response::new(ListUsersResponse {
            users,
            next_page_token: "next_token".to_string()
        }))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "0.0.0.0:50051".parse()?;
    let user_service = MyUserService::default();

    Server::builder()
        .add_service(UserServiceServer::new(user_service))
        .serve(addr)
        .await?;

    Ok(())
}

Server Streaming

use tokio_stream::wrappers::ReceiverStream;
use tokio::sync::mpsc;

impl LogService for MyLogService {
    type StreamLogsStream = ReceiverStream<Result<LogEntry, Status>>;

    async fn stream_logs(
        &self,
        request: Request<StreamLogsRequest>
    ) -> Result<Response<Self::StreamLogsStream>, Status> {
        let (tx, rx) = mpsc::channel(128);

        tokio::spawn(async move {
            for i in 0..10 {
                let log = LogEntry {
                    timestamp: Some(prost_types::Timestamp::from(SystemTime::now())),
                    level: "INFO".to_string(),
                    message: format!("Log entry {}", i)
                };

                tx.send(Ok(log)).await.unwrap();
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        });

        Ok(Response::new(ReceiverStream::new(rx)))
    }
}

Client Implementation

use user::user_service_client::UserServiceClient;
use user::GetUserRequest;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = UserServiceClient::connect("http://[::1]:50051").await?;

    let request = tonic::Request::new(GetUserRequest {
        id: "123".to_string()
    });

    let response = client.get_user(request).await?;

    println!("User: {:?}", response.into_inner().user);

    Ok(())
}

Go: Connect-Go

Connect-Go provides gRPC-compatible APIs that work in browsers.

Installation

go install connectrpc.com/connect/cmd/protoc-gen-connect-go@latest

Code Generation

protoc --go_out=. --go_opt=paths=source_relative \
       --connect-go_out=. --connect-go_opt=paths=source_relative \
       user/v1/user.proto

Server Implementation

package main

import (
    "context"
    "net/http"

    "connectrpc.com/connect"
    userv1 "example.com/user/v1"
    "example.com/user/v1/userv1connect"
)

type UserServiceHandler struct{}

func (s *UserServiceHandler) GetUser(
    ctx context.Context,
    req *connect.Request[userv1.GetUserRequest],
) (*connect.Response[userv1.GetUserResponse], error) {
    user, err := getUserFromDB(req.Msg.Id)
    if err != nil {
        return nil, connect.NewError(connect.CodeNotFound, err)
    }

    return connect.NewResponse(&userv1.GetUserResponse{
        User: user,
    }), nil
}

func main() {
    handler := &UserServiceHandler{}
    path, handlerFunc := userv1connect.NewUserServiceHandler(handler)

    http.Handle(path, handlerFunc)
    http.ListenAndServe(":8080", nil)
}

Client (Browser-Compatible)

import { createPromiseClient } from '@connectrpc/connect'
import { createConnectTransport } from '@connectrpc/connect-web'
import { UserService } from './gen/user/v1/user_connect'

const transport = createConnectTransport({
  baseUrl: 'http://localhost:8080'
})

const client = createPromiseClient(UserService, transport)

async function getUser(id: string) {
  const response = await client.getUser({ id })
  console.log(response.user)
}

Python: grpcio

Installation

pip install grpcio grpcio-tools

Code Generation

python -m grpc_tools.protoc \
    -I./proto \
    --python_out=. \
    --grpc_python_out=. \
    proto/user.proto

Server Implementation

import grpc
from concurrent import futures
import user_pb2
import user_pb2_grpc

class UserServiceServicer(user_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        user = get_user_from_db(request.id)

        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details('User not found')
            return user_pb2.GetUserResponse()

        return user_pb2.GetUserResponse(user=user)

    def ListUsers(self, request, context):
        users = get_users_from_db(
            page_size=request.page_size,
            page_token=request.page_token
        )

        return user_pb2.ListUsersResponse(
            users=users,
            next_page_token="next_token"
        )

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    user_pb2_grpc.add_UserServiceServicer_to_server(
        UserServiceServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

Client Implementation

import grpc
import user_pb2
import user_pb2_grpc

def run():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)

        response = stub.GetUser(user_pb2.GetUserRequest(id='123'))
        print(f"User: {response.user}")

if __name__ == '__main__':
    run()

Error Handling

gRPC Status Codes

Rust Error Handling

use tonic::{Status, Code};

impl UserService for MyUserService {
    async fn get_user(
        &self,
        request: Request<GetUserRequest>
    ) -> Result<Response<GetUserResponse>, Status> {
        let user = get_user_from_db(&request.into_inner().id)
            .await
            .map_err(|e| Status::new(Code::NotFound, format!("User not found: {}", e)))?;

        Ok(Response::new(GetUserResponse { user: Some(user) }))
    }
}

Best Practices

1. Use proto3 syntax - Latest version with better defaults 2. Version your services - Use package names like user.v1, user.v2 3. Reserve field numbers - When removing fields, mark as reserved 4. Use streaming for large data - Don't send large responses in single message 5. Implement health checks - Use gRPC health checking protocol 6. Enable reflection - For debugging with tools like grpcurl 7. Use interceptors - For logging, auth, metrics 8. Handle backpressure - Limit concurrent requests 9. Set timeouts - Prevent hanging connections 10. Document your services - Comments in proto files

Performance Tips

Use HTTP/2 multiplexing:

• Single connection for multiple RPCs
• Reduces connection overhead

Enable compression:

Server::builder()
    .add_service(
        UserServiceServer::new(service)
            .send_compressed(CompressionEncoding::Gzip)
            .accept_compressed(CompressionEncoding::Gzip)
    )

Connection pooling:

• Reuse gRPC channels
• Don't create new channel per request

Batch requests when possible:

• Use repeated fields for batch operations
• Reduces network round trips

OpenAPI Documentation Guide

Overview

OpenAPI (formerly Swagger) provides a standard way to document REST APIs. This guide covers automatic generation with FastAPI, Hono, Axum, and Gin.

Table of Contents

• OpenAPI 3.1 Basics
• FastAPI (Automatic)
• Hono (Middleware)
• Axum (utoipa)
• Gin (swaggo/swag)
• Interactive Documentation

OpenAPI 3.1 Basics

Minimal OpenAPI Document

openapi: 3.1.0
info:
  title: My API
  version: 1.0.0
  description: API for managing users and posts

servers:
  - url: https://api.example.com/v1
    description: Production
  - url: http://localhost:3000/v1
    description: Development

paths:
  /users:
    get:
      summary: List users
      operationId: listUsers
      tags:
        - users
      parameters:
        - name: limit
          in: query
          schema:
            type: integer
            default: 20
      responses:
        '200':
          description: List of users
          content:
            application/json:
              schema:
                type: object
                properties:
                  users:
                    type: array
                    items:
                      $ref: '#/components/schemas/User'

components:
  schemas:
    User:
      type: object
      required:
        - id
        - name
        - email
      properties:
        id:
          type: string
        name:
          type: string
        email:
          type: string
          format: email

FastAPI (Automatic)

FastAPI generates OpenAPI automatically from Python type hints.

Zero Configuration

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI(
    title="My API",
    description="API for managing users",
    version="1.0.0",
    openapi_tags=[
        {"name": "users", "description": "User operations"},
        {"name": "posts", "description": "Post operations"}
    ]
)

class User(BaseModel):
    """User model"""
    name: str
    email: EmailStr
    age: int | None = None

@app.get(
    "/users",
    tags=["users"],
    summary="List all users",
    response_description="List of users"
)
async def list_users(
    limit: int = 20,
    offset: int = 0
) -> list[User]:
    """
    Retrieve a list of users.

    - **limit**: Maximum number of users to return
    - **offset**: Number of users to skip
    """
    return get_users_from_db(limit, offset)

@app.post(
    "/users",
    tags=["users"],
    status_code=201,
    response_model=User
)
async def create_user(user: User) -> User:
    """Create a new user"""
    return create_user_in_db(user)

# Docs automatically available at:
# - /docs (Swagger UI)
# - /redoc (ReDoc)
# - /openapi.json (OpenAPI spec)

Advanced Configuration

from fastapi.openapi.utils import get_openapi

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

    openapi_schema = get_openapi(
        title="My API",
        version="1.0.0",
        description="Comprehensive API documentation",
        routes=app.routes,
    )

    # Add custom fields
    openapi_schema["info"]["x-logo"] = {
        "url": "https://example.com/logo.png"
    }

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

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

Response Models

from pydantic import BaseModel

class ErrorResponse(BaseModel):
    detail: str

class UserResponse(BaseModel):
    user: User

@app.get(
    "/users/{user_id}",
    response_model=UserResponse,
    responses={
        404: {
            "model": ErrorResponse,
            "description": "User not found"
        },
        500: {
            "model": ErrorResponse,
            "description": "Internal server error"
        }
    }
)
async def get_user(user_id: str):
    user = get_user_from_db(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"user": user}

Hono (Middleware)

Installation

npm install @hono/zod-openapi

Setup

import { OpenAPIHono, createRoute, z } from '@hono/zod-openapi'

const app = new OpenAPIHono()

const UserSchema = z.object({
  id: z.string().openapi({ example: '123' }),
  name: z.string().openapi({ example: 'Alice' }),
  email: z.string().email().openapi({ example: 'alice@example.com' })
})

const getUserRoute = createRoute({
  method: 'get',
  path: '/users/{id}',
  request: {
    params: z.object({
      id: z.string()
    })
  },
  responses: {
    200: {
      content: {
        'application/json': {
          schema: UserSchema
        }
      },
      description: 'User found'
    },
    404: {
      description: 'User not found'
    }
  },
  tags: ['users'],
  summary: 'Get user by ID'
})

app.openapi(getUserRoute, (c) => {
  const { id } = c.req.valid('param')
  const user = getUserFromDB(id)

  if (!user) {
    return c.json({ error: 'Not found' }, 404)
  }

  return c.json(user, 200)
})

// Serve OpenAPI spec
app.doc('/openapi.json', {
  openapi: '3.1.0',
  info: {
    title: 'My API',
    version: '1.0.0'
  }
})

// Swagger UI
import { swaggerUI } from '@hono/swagger-ui'
app.get('/docs', swaggerUI({ url: '/openapi.json' }))

export default app

Axum (utoipa)

Installation

[dependencies]
utoipa = { version = "4", features = ["axum_extras"] }
utoipa-swagger-ui = { version = "6", features = ["axum"] }

Implementation

use axum::{Router, Json};
use utoipa::{OpenApi, ToSchema};
use utoipa_swagger_ui::SwaggerUi;

#[derive(serde::Serialize, ToSchema)]
struct User {
    #[schema(example = "123")]
    id: String,
    #[schema(example = "Alice")]
    name: String,
    #[schema(example = "alice@example.com")]
    email: String,
}

#[derive(OpenApi)]
#[openapi(
    paths(
        list_users,
        get_user,
        create_user
    ),
    components(
        schemas(User)
    ),
    tags(
        (name = "users", description = "User management endpoints")
    )
)]
struct ApiDoc;

/// List all users
#[utoipa::path(
    get,
    path = "/users",
    tag = "users",
    responses(
        (status = 200, description = "List of users", body = Vec<User>)
    )
)]
async fn list_users() -> Json<Vec<User>> {
    Json(get_users_from_db())
}

/// Get user by ID
#[utoipa::path(
    get,
    path = "/users/{id}",
    tag = "users",
    params(
        ("id" = String, Path, description = "User ID")
    ),
    responses(
        (status = 200, description = "User found", body = User),
        (status = 404, description = "User not found")
    )
)]
async fn get_user(
    axum::extract::Path(id): axum::extract::Path<String>
) -> Result<Json<User>, StatusCode> {
    match get_user_from_db(&id) {
        Some(user) => Ok(Json(user)),
        None => Err(StatusCode::NOT_FOUND)
    }
}

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/users", get(list_users))
        .route("/users/:id", get(get_user))
        .merge(SwaggerUi::new("/docs").url("/openapi.json", ApiDoc::openapi()));

    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

Gin (swaggo/swag)

Installation

go install github.com/swaggo/swag/cmd/swag@latest

Implementation

package main

import (
    "github.com/gin-gonic/gin"
    swaggerFiles "github.com/swaggo/files"
    ginSwagger "github.com/swaggo/gin-swagger"

    _ "example.com/docs" // Generated by swag init
)

type User struct {
    ID    string `json:"id" example:"123"`
    Name  string `json:"name" example:"Alice"`
    Email string `json:"email" example:"alice@example.com"`
}

// @title My API
// @version 1.0
// @description API for managing users
// @host localhost:8080
// @BasePath /api/v1
func main() {
    r := gin.Default()

    v1 := r.Group("/api/v1")
    {
        v1.GET("/users", listUsers)
        v1.GET("/users/:id", getUser)
        v1.POST("/users", createUser)
    }

    // Swagger docs
    r.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))

    r.Run(":8080")
}

// @Summary List users
// @Description Get list of all users
// @Tags users
// @Accept json
// @Produce json
// @Param limit query int false "Limit" default(20)
// @Success 200 {array} User
// @Router /users [get]
func listUsers(c *gin.Context) {
    users := getUsersFromDB()
    c.JSON(200, users)
}

// @Summary Get user
// @Description Get user by ID
// @Tags users
// @Accept json
// @Produce json
// @Param id path string true "User ID"
// @Success 200 {object} User
// @Failure 404 {object} ErrorResponse
// @Router /users/{id} [get]
func getUser(c *gin.Context) {
    id := c.Param("id")
    user, exists := getUserFromDB(id)

    if !exists {
        c.JSON(404, gin.H{"error": "User not found"})
        return
    }

    c.JSON(200, user)
}

// @Summary Create user
// @Description Create a new user
// @Tags users
// @Accept json
// @Produce json
// @Param user body User true "User"
// @Success 201 {object} User
// @Failure 400 {object} ErrorResponse
// @Router /users [post]
func createUser(c *gin.Context) {
    var user User
    if err := c.ShouldBindJSON(&user); err != nil {
        c.JSON(400, gin.H{"error": err.Error()})
        return
    }

    created := createUserInDB(user)
    c.JSON(201, created)
}

Generate Docs

swag init
# Creates docs/ directory with swagger.json and swagger.yaml

Interactive Documentation

Swagger UI Customization

from fastapi import FastAPI
from fastapi.openapi.docs import get_swagger_ui_html

app = FastAPI(docs_url=None, redoc_url=None)

@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
    return get_swagger_ui_html(
        openapi_url="/openapi.json",
        title=app.title + " - Swagger UI",
        oauth2_redirect_url="/docs/oauth2-redirect",
        swagger_js_url="https://unpkg.com/swagger-ui-dist@5/swagger-ui-bundle.js",
        swagger_css_url="https://unpkg.com/swagger-ui-dist@5/swagger-ui.css",
        swagger_ui_parameters={"persistAuthorization": True}
    )

ReDoc Customization

from fastapi.openapi.docs import get_redoc_html

@app.get("/redoc", include_in_schema=False)
async def redoc_html():
    return get_redoc_html(
        openapi_url="/openapi.json",
        title=app.title + " - ReDoc",
        redoc_js_url="https://unpkg.com/redoc@latest/bundles/redoc.standalone.js",
    )

Best Practices

1. Use semantic versioning - Version your API (v1, v2) 2. Document all endpoints - Include descriptions and examples 3. Define schemas - Reuse schema definitions 4. Include examples - Real-world request/response examples 5. Document error responses - All possible error codes 6. Add security schemes - Document authentication methods 7. Tag endpoints - Group related endpoints 8. Include server URLs - Production, staging, development 9. Keep docs in sync - Auto-generation preferred 10. Test with Swagger UI - Verify docs are usable

Pagination Patterns

Overview

Pagination divides large datasets into manageable chunks. This guide covers cursor-based, offset-based, and keyset pagination with implementation examples.

Table of Contents

• Cursor-Based Pagination
• Offset-Based Pagination
• Keyset Pagination
• Comparison Matrix
• Frontend Integration

Cursor-Based Pagination

Advantages

• Scales to billions of records: O(1) lookup regardless of dataset size
• Handles real-time changes: No skipped or duplicate records when data changes
• Consistent performance: No degradation at high page numbers
• Works with any sortable column: Not limited to IDs

Disadvantages

• No direct page access: Can't jump to arbitrary page number
• Stateless cursors required: Cursor must encode position completely
• More complex implementation: Requires cursor encoding/decoding

When to Use

✅ Use cursor pagination when:

• Dataset changes frequently (real-time data)
• Dataset is large (>10,000 records)
• Consistency is critical (no duplicates/skips acceptable)
• Infinite scroll UX pattern
• API will scale to millions of records

❌ Avoid cursor pagination when:

• Dataset is small and static
• Direct page number access required
• Implementation complexity not justified

Implementation (FastAPI)

from fastapi import Query
from typing import Optional
from pydantic import BaseModel

class PaginatedResponse(BaseModel):
    items: list
    next_cursor: Optional[str]
    has_more: bool

@app.get("/items", response_model=PaginatedResponse)
async def list_items(
    cursor: Optional[str] = Query(None, description="Pagination cursor"),
    limit: int = Query(20, ge=1, le=100, description="Items per page")
):
    query = db.query(Item).order_by(Item.id)

    # Apply cursor filter
    if cursor:
        query = query.filter(Item.id > cursor)

    # Fetch limit + 1 to check if more results exist
    items = query.limit(limit + 1).all()

    # Check if there are more results
    has_more = len(items) > limit
    if has_more:
        items = items[:limit]

    # Next cursor is the last item's ID
    next_cursor = items[-1].id if items else None

    return {
        "items": [item.to_dict() for item in items],
        "next_cursor": next_cursor,
        "has_more": has_more
    }

Implementation (Hono)

import { Hono } from 'hono'

const app = new Hono()

app.get('/items', async (c) => {
  const cursor = c.req.query('cursor')
  const limit = parseInt(c.req.query('limit') || '20')

  let query = db.select().from(items).orderBy(items.id)

  if (cursor) {
    query = query.where(gt(items.id, cursor))
  }

  const results = await query.limit(limit + 1)

  const hasMore = results.length > limit
  const items = hasMore ? results.slice(0, limit) : results
  const nextCursor = items.length > 0 ? items[items.length - 1].id : null

  return c.json({
    items,
    next_cursor: nextCursor,
    has_more: hasMore
  })
})

Cursor Encoding (Opaque Cursors)

For security and flexibility, encode cursor as base64 JSON:

import base64
import json

def encode_cursor(item_id: str, timestamp: str) -> str:
    """Encode cursor with multiple fields"""
    cursor_data = {
        "id": item_id,
        "timestamp": timestamp
    }
    json_str = json.dumps(cursor_data)
    return base64.b64encode(json_str.encode()).decode()

def decode_cursor(cursor: str) -> dict:
    """Decode cursor back to fields"""
    json_str = base64.b64decode(cursor.encode()).decode()
    return json.loads(json_str)

@app.get("/items")
async def list_items(cursor: Optional[str] = None, limit: int = 20):
    query = db.query(Item).order_by(Item.created_at, Item.id)

    if cursor:
        cursor_data = decode_cursor(cursor)
        query = query.filter(
            (Item.created_at > cursor_data["timestamp"]) |
            ((Item.created_at == cursor_data["timestamp"]) & (Item.id > cursor_data["id"]))
        )

    items = query.limit(limit + 1).all()
    has_more = len(items) > limit
    if has_more:
        items = items[:limit]

    next_cursor = None
    if items:
        last_item = items[-1]
        next_cursor = encode_cursor(last_item.id, last_item.created_at)

    return {
        "items": [item.to_dict() for item in items],
        "next_cursor": next_cursor,
        "has_more": has_more
    }

Multi-Column Sorting

For sorting by multiple columns (e.g., ORDER BY created_at DESC, id ASC):

from sqlalchemy import and_, or_

@app.get("/items")
async def list_items(cursor: Optional[str] = None, limit: int = 20):
    query = db.query(Item).order_by(Item.created_at.desc(), Item.id.asc())

    if cursor:
        cursor_data = decode_cursor(cursor)
        # WHERE (created_at < cursor.created_at) OR
        #       (created_at = cursor.created_at AND id > cursor.id)
        query = query.filter(
            or_(
                Item.created_at < cursor_data["timestamp"],
                and_(
                    Item.created_at == cursor_data["timestamp"],
                    Item.id > cursor_data["id"]
                )
            )
        )

    items = query.limit(limit + 1).all()
    # ... rest of implementation

Offset-Based Pagination

Advantages

• Simple to implement: Just OFFSET and LIMIT in SQL
• Direct page access: Can jump to any page number
• Easy to understand: Familiar "page 1, 2, 3..." UX

Disadvantages

• Poor performance at high offsets: OFFSET 1000000 scans 1M rows
• Inconsistent results: Skipped/duplicate records when data changes
• Not scalable: Performance degrades with dataset size

When to Use

✅ Use offset pagination when:

• Dataset is small (<10,000 records)
• Dataset is static (rarely changes)
• Direct page number access required
• Implementation simplicity critical

❌ Avoid offset pagination when:

• Dataset is large or growing
• Data changes frequently
• Performance at scale matters

Implementation (FastAPI)

from math import ceil

@app.get("/items")
async def list_items(
    page: int = Query(1, ge=1, description="Page number"),
    per_page: int = Query(20, ge=1, le=100, description="Items per page")
):
    # Calculate offset
    offset = (page - 1) * per_page

    # Query items
    items = db.query(Item).offset(offset).limit(per_page).all()

    # Get total count (expensive!)
    total = db.query(Item).count()

    return {
        "items": [item.to_dict() for item in items],
        "page": page,
        "per_page": per_page,
        "total": total,
        "total_pages": ceil(total / per_page),
        "has_next": page < ceil(total / per_page),
        "has_prev": page > 1
    }

Performance Optimization

Problem: High offset scans many rows:

SELECT * FROM items OFFSET 1000000 LIMIT 20;
-- Scans 1,000,020 rows to return 20!

Solution: Use keyset pagination (see below) or limit maximum offset:

MAX_OFFSET = 10000  # Limit to first 10,000 records

@app.get("/items")
async def list_items(page: int = 1, per_page: int = 20):
    offset = (page - 1) * per_page

    if offset > MAX_OFFSET:
        raise HTTPException(
            status_code=400,
            detail=f"Maximum offset is {MAX_OFFSET}. Use cursor pagination for deep pagination."
        )

    # ... rest of implementation

Keyset Pagination

Keyset pagination is a variant of cursor-based pagination using the last seen value directly in the query.

Implementation

@app.get("/items")
async def list_items(
    last_id: Optional[int] = Query(None, description="ID of last item from previous page"),
    limit: int = Query(20, le=100)
):
    query = db.query(Item).order_by(Item.id)

    if last_id:
        query = query.filter(Item.id > last_id)

    items = query.limit(limit).all()

    return {
        "items": [item.to_dict() for item in items],
        "last_id": items[-1].id if items else None
    }

Advantages over cursor-based:

• Simpler (no encoding/decoding)
• Transparent to client

Disadvantages:

• Exposes database IDs
• Harder to change sort logic
• Can't sort by multiple columns easily

Recommendation: Use cursor-based with encoded cursors for production. Use keyset for simple internal APIs.

Comparison Matrix

Frontend Integration

React + Cursor Pagination

import { useState, useEffect } from 'react'

interface PaginatedResponse<T> {
  items: T[]
  next_cursor: string | null
  has_more: boolean
}

function useInfiniteScroll<T>(endpoint: string) {
  const [items, setItems] = useState<T[]>([])
  const [cursor, setCursor] = useState<string | null>(null)
  const [loading, setLoading] = useState(false)
  const [hasMore, setHasMore] = useState(true)

  const loadMore = async () => {
    if (loading || !hasMore) return

    setLoading(true)

    const url = cursor
      ? `${endpoint}?cursor=${cursor}&limit=20`
      : `${endpoint}?limit=20`

    const response = await fetch(url)
    const data: PaginatedResponse<T> = await response.json()

    setItems(prev => [...prev, ...data.items])
    setCursor(data.next_cursor)
    setHasMore(data.has_more)
    setLoading(false)
  }

  return { items, loadMore, loading, hasMore }
}

// Usage
function ItemList() {
  const { items, loadMore, loading, hasMore } = useInfiniteScroll('/api/items')

  return (
    <div>
      {items.map(item => (
        <div key={item.id}>{item.name}</div>
      ))}

      {hasMore && (
        <button onClick={loadMore} disabled={loading}>
          {loading ? 'Loading...' : 'Load More'}
        </button>
      )}
    </div>
  )
}

React + Offset Pagination

import { useState, useEffect } from 'react'

interface OffsetPaginatedResponse<T> {
  items: T[]
  page: number
  per_page: number
  total: number
  total_pages: number
}

function usePagination<T>(endpoint: string) {
  const [items, setItems] = useState<T[]>([])
  const [page, setPage] = useState(1)
  const [totalPages, setTotalPages] = useState(1)
  const [loading, setLoading] = useState(false)

  const loadPage = async (pageNum: number) => {
    setLoading(true)

    const response = await fetch(`${endpoint}?page=${pageNum}&per_page=20`)
    const data: OffsetPaginatedResponse<T> = await response.json()

    setItems(data.items)
    setPage(data.page)
    setTotalPages(data.total_pages)
    setLoading(false)
  }

  useEffect(() => {
    loadPage(page)
  }, [page])

  return {
    items,
    page,
    totalPages,
    loading,
    goToPage: setPage,
    nextPage: () => setPage(p => Math.min(p + 1, totalPages)),
    prevPage: () => setPage(p => Math.max(p - 1, 1))
  }
}

// Usage
function ItemList() {
  const { items, page, totalPages, loading, goToPage, nextPage, prevPage } =
    usePagination('/api/items')

  return (
    <div>
      {items.map(item => (
        <div key={item.id}>{item.name}</div>
      ))}

      <div>
        <button onClick={prevPage} disabled={page === 1 || loading}>
          Previous
        </button>

        <span>Page {page} of {totalPages}</span>

        <button onClick={nextPage} disabled={page === totalPages || loading}>
          Next
        </button>
      </div>
    </div>
  )
}

Integration with TanStack Table

See tables skill for integration with table virtualization and sorting.

import { useInfiniteQuery } from '@tanstack/react-query'
import { useVirtualizer } from '@tanstack/react-virtual'

async function fetchItems({ pageParam = null }) {
  const url = pageParam
    ? `/api/items?cursor=${pageParam}&limit=50`
    : '/api/items?limit=50'

  const response = await fetch(url)
  return response.json()
}

function VirtualizedTable() {
  const {
    data,
    fetchNextPage,
    hasNextPage,
    isFetchingNextPage
  } = useInfiniteQuery({
    queryKey: ['items'],
    queryFn: fetchItems,
    getNextPageParam: (lastPage) => lastPage.next_cursor
  })

  const allItems = data?.pages.flatMap(page => page.items) ?? []

  // Virtualization setup
  const parentRef = useRef<HTMLDivElement>(null)

  const virtualizer = useVirtualizer({
    count: allItems.length,
    getScrollElement: () => parentRef.current,
    estimateSize: () => 50,
    overscan: 5
  })

  // Load more when scrolled near bottom
  useEffect(() => {
    const [lastItem] = [...virtualizer.getVirtualItems()].reverse()

    if (
      lastItem &&
      lastItem.index >= allItems.length - 1 &&
      hasNextPage &&
      !isFetchingNextPage
    ) {
      fetchNextPage()
    }
  }, [virtualizer.getVirtualItems(), hasNextPage, isFetchingNextPage])

  return (
    <div ref={parentRef} style={{ height: '600px', overflow: 'auto' }}>
      <div style={{ height: `${virtualizer.getTotalSize()}px`, position: 'relative' }}>
        {virtualizer.getVirtualItems().map(virtualRow => {
          const item = allItems[virtualRow.index]
          return (
            <div
              key={virtualRow.key}
              style={{
                position: 'absolute',
                top: 0,
                left: 0,
                width: '100%',
                height: `${virtualRow.size}px`,
                transform: `translateY(${virtualRow.start}px)`
              }}
            >
              {item.name}
            </div>
          )
        })}
      </div>
    </div>
  )
}

Database Indexing

For optimal pagination performance, ensure proper indexing:

-- Cursor pagination on ID
CREATE INDEX idx_items_id ON items(id);

-- Cursor pagination on created_at + id
CREATE INDEX idx_items_created_id ON items(created_at DESC, id ASC);

-- Offset pagination (less important but helpful)
CREATE INDEX idx_items_id ON items(id);

Best Practices Summary

1. Use cursor pagination for production APIs - Scales better, handles real-time data 2. Encode cursors as opaque tokens - Security and flexibility 3. Return `has_more` flag - Avoids extra count query 4. Limit maximum page size - Prevent resource exhaustion (max 100 items) 5. Index sort columns - Essential for performance 6. Document pagination in OpenAPI - Include cursor/page parameters 7. Consistent ordering - Always include tiebreaker (ID) for deterministic results 8. Cache count queries - For offset pagination, cache total count 9. *Avoid `COUNT()` when possible - Expensive on large tables 10. Use virtualization on frontend** - For smooth infinite scroll experience

Migration Path

From offset to cursor pagination:

1. Support both simultaneously with different endpoints or parameters 2. Deprecate offset version over time 3. Provide migration guide for API consumers

@app.get("/items")
async def list_items(
    # Cursor-based (new)
    cursor: Optional[str] = None,
    # Offset-based (deprecated)
    page: Optional[int] = None,
    per_page: int = 20
):
    if page is not None:
        # Offset-based (deprecated)
        warnings.warn("Offset pagination is deprecated. Use cursor parameter.", DeprecationWarning)
        return offset_paginate(page, per_page)
    else:
        # Cursor-based (recommended)
        return cursor_paginate(cursor, per_page)