Seven API Optimization Techniques That Actually Work

from fastapi import FastAPI
from fastapi.responses import JSONResponse

@app.get("/api/products/{product_id}")
async def get_product(product_id: int):
    product = await db.fetch_product(product_id)
    return JSONResponse(
        content=product,
        headers={
            "Cache-Control": "public, max-age=3600",  # Cache for 1 hour
            "ETag": compute_etag(product),
        }
    )

# Bad: N+1 queries
@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    user = await db.fetch_user(user_id)
    user.orders = await db.fetch_orders(user.id)  # Extra query
    return user

# Good: Single query with JOIN
@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    user = await db.fetch("""
        SELECT u.*, json_agg(o.*) as orders
        FROM users u
        LEFT JOIN orders o ON u.id = o.user_id
        WHERE u.id = $1
        GROUP BY u.id
    """, user_id)
    return user

import redis
import json
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cache_result(ttl=3600):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Create cache key from function name and arguments
            cache_key = f"{func.__name__}:{args}:{kwargs}"
            
            # Try to get from cache
            cached = redis_client.get(cache_key)
            if cached:
                return json.loads(cached)
            
            # Compute and cache
            result = await func(*args, **kwargs)
            redis_client.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@app.get("/api/recommendations/{user_id}")
@cache_result(ttl=1800)
async def get_recommendations(user_id: int):
    # Expensive recommendation algorithm
    return compute_recommendations(user_id)

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    "postgresql://user:pass@localhost/db",
    poolclass=QueuePool,
    pool_size=20,           # Number of connections to keep open
    max_overflow=10,        # Additional connections if needed
    pool_recycle=3600,      # Recycle connections after 1 hour
    pool_pre_ping=True,     # Test connections before use
)

# Use engine to get connections from pool
async def get_db():
    async with engine.connect() as conn:
        yield conn

from fastapi.middleware.gzip import GZIPMiddleware

app.add_middleware(GZIPMiddleware, minimum_size=1000)

from datetime import datetime, timedelta

# Store writes in a buffer
write_buffer = []
write_lock = asyncio.Lock()

@app.post("/api/events")
async def log_event(event: Event):
    async with write_lock:
        write_buffer.append(event)
    return {"status": "queued"}

async def flush_buffer():
    while True:
        await asyncio.sleep(5)  # Flush every 5 seconds
        
        async with write_lock:
            if not write_buffer:
                continue
            
            # Insert 1000 events in one query instead of 1000 queries
            await db.execute(
                "INSERT INTO events (type, data) VALUES (unnest($1), unnest($2))",
                [e.type for e in write_buffer],
                [json.dumps(e.data) for e in write_buffer],
            )
            write_buffer.clear()

# Start the flush task
@app.on_event("startup")
async def startup():
    asyncio.create_task(flush_buffer())

from pydantic import BaseModel

class PaginationParams(BaseModel):
    limit: int = 20
    offset: int = 0
    
    @validator('limit')
    def limit_max_100(cls, v):
        if v > 100:
            raise ValueError('limit cannot exceed 100')
        return v

@app.get("/api/posts")
async def list_posts(params: PaginationParams = Depends()):
    posts = await db.fetch(
        "SELECT * FROM posts ORDER BY created_at DESC LIMIT $1 OFFSET $2",
        params.limit,
        params.offset,
    )
    return posts

from time import time

@app.middleware("http")
async def add_timing(request, call_next):
    start = time()
    response = await call_next(request)
    duration = time() - start
    
    response.headers["X-Process-Time"] = str(duration)
    
    # Log slow requests
    if duration > 0.5:
        logger.warning(f"{request.url.path} took {duration:.2f}s")
    
    return response