Production-Ready Code Examples

Antonio Delgado — Sun, 07 Dec 2025 22:07:53 +0000

Production-Ready Code Examples

I'll upgrade all code examples to production quality with realistic naming, error handling, comments, and the evolution real developers go through.

The Database Query That Still Haunts Me

Version 1: What I Actually Shipped (The Bad One)

// user-filters.controller.js
// Added by @mike - Sprint 23 - User filtering feature
// TODO: This works but feels slow with lots of filters, revisit if issues

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

const prisma = new PrismaClient();

/**
 * GET /api/users/search
 * Filters users by multiple criteria
 */
export async function searchUsers(req, res) {
  const { filters } = req.body; // filters = [{ field: 'email', value: 'gmail' }, ...]

  try {
    // Copilot suggested this pattern - seemed clean
    const users = await prisma.user.findMany({
      where: {
        OR: filters.map(filter => ({
          [filter.field]: { contains: filter.value }
        }))
      }
    });

    return res.json({ users, count: users.length });
  } catch (error) {
    // FIXME: Error handling is too generic
    console.error('Search failed:', error);
    return res.status(500).json({ error: 'Search failed' });
  }
}

What's wrong:

No validation on filters input
contains does case-sensitive search (users complained about this)
Full table scan for each OR condition
No pagination
Error logging with console.error instead of proper logger
Returns all user data including sensitive fields

Version 2: After Performance Issues Hit Production

// user-filters.controller.js
// Updated by @mike - Sprint 26 - Performance fixes after incident #2847
// Context: Dashboard timing out with >100 users, support tickets piling up

import { PrismaClient } from '@prisma/client';
import { logger } from '../utils/logger.js';
import { ValidationError } from '../errors/index.js';

const prisma = new PrismaClient();

// Allowed fields to prevent SQL injection via dynamic field names
const SEARCHABLE_FIELDS = new Set([
  'email', 
  'username', 
  'firstName', 
  'lastName', 
  'department'
]);

/**
 * GET /api/users/search
 * Filters users by multiple criteria with pagination
 * 
 * @param {Array} filters - Array of {field, value} objects
 * @param {Number} page - Page number (default: 1)
 * @param {Number} limit - Results per page (default: 50, max: 100)
 */
export async function searchUsers(req, res) {
  try {
    const { filters = [], page = 1, limit = 50 } = req.body;

    // Input validation - learned this the hard way
    if (!Array.isArray(filters)) {
      throw new ValidationError('filters must be an array');
    }

    if (filters.length === 0) {
      return res.json({ users: [], count: 0, page, totalPages: 0 });
    }

    if (filters.length > 10) {
      throw new ValidationError('Maximum 10 filters allowed');
    }

    // Validate each filter
    const validatedFilters = filters.map((filter, index) => {
      if (!filter.field || !filter.value) {
        throw new ValidationError(
          `Filter at index ${index} missing required fields`
        );
      }

      if (!SEARCHABLE_FIELDS.has(filter.field)) {
        throw new ValidationError(
          `Field "${filter.field}" is not searchable. ` +
          `Allowed fields: ${Array.from(SEARCHABLE_FIELDS).join(', ')}`
        );
      }

      return {
        field: filter.field,
        value: String(filter.value).trim()
      };
    });

    // Build where clause - this is the part that actually got fixed
    // Old way: OR with contains on each field = disaster
    // New way: Compound AND with proper indexes
    const whereConditions = validatedFilters.reduce((conditions, filter) => {
      // Case-insensitive search - users kept complaining about this
      conditions[filter.field] = {
        contains: filter.value,
        mode: 'insensitive'
      };
      return conditions;
    }, {});

    // Get total count for pagination (separate query, but cached)
    const totalCount = await prisma.user.count({
      where: whereConditions
    });

    const totalPages = Math.ceil(totalCount / Math.min(limit, 100));
    const offset = (page - 1) * limit;

    // Actual query with proper pagination and field selection
    const users = await prisma.user.findMany({
      where: whereConditions,
      select: {
        id: true,
        email: true,
        username: true,
        firstName: true,
        lastName: true,
        department: true,
        createdAt: true,
        // Don't return password hash, lastLoginIp, etc.
      },
      take: Math.min(limit, 100), // Hard cap at 100
      skip: offset,
      orderBy: {
        createdAt: 'desc'
      }
    });

    // Log for debugging slow queries
    if (users.length > 0) {
      logger.info('User search completed', {
        filterCount: validatedFilters.length,
        resultCount: users.length,
        page,
        executionTime: `${Date.now() - req.startTime}ms` // Added in middleware
      });
    }

    return res.json({
      users,
      count: users.length,
      totalCount,
      page: Number(page),
      totalPages,
      hasMore: page < totalPages
    });

  } catch (error) {
    if (error instanceof ValidationError) {
      return res.status(400).json({ 
        error: error.message,
        code: 'VALIDATION_ERROR'
      });
    }

    // Log actual errors for debugging
    logger.error('User search failed', {
      error: error.message,
      stack: error.stack,
      filters: req.body.filters,
      userId: req.user?.id // Added after auth middleware
    });

    return res.status(500).json({ 
      error: 'An error occurred while searching users',
      code: 'SEARCH_ERROR'
    });
  }
}

Migration required:

-- Added after performance incident
-- Reduced query time from ~3000ms to ~50ms with 50k users

CREATE INDEX idx_user_email_trgm ON users USING gin (email gin_trgm_ops);
CREATE INDEX idx_user_username_trgm ON users USING gin (username gin_trgm_ops);
CREATE INDEX idx_user_firstname_trgm ON users USING gin (first_name gin_trgm_ops);
CREATE INDEX idx_user_lastname_trgm ON users USING gin (last_name gin_trgm_ops);
CREATE INDEX idx_user_department ON users (department);

-- Enable trigram extension for fuzzy matching
CREATE EXTENSION IF NOT EXISTS pg_trgm;

.cursorrules for FastAPI Projects (Real Production Version)

# .cursorrules
# FastAPI Backend Rules - Updated 2024-01-15
# Team: Backend Squad | Maintainer: @sarah
# 
# This file teaches Cursor our conventions so we stop fixing the same
# things in every PR. If you update this, ping #backend-squad.

## Tech Stack & Versions
# We're locked to these versions until Q2 upgrade sprint
- Python 3.11.7
- fastapi==0.104.1
- pydantic==2.5.0
- sqlalchemy==2.0.23
- alembic==1.13.0
- pytest==7.4.3
- pytest-asyncio==0.21.1
- redis==5.0.1
- celery==5.3.4

## Code Style (Non-Negotiable)
- Type hints are MANDATORY. No exceptions. PR will be rejected.
- Use Pydantic v2 for all request/response models (not v1 syntax!)
- Async/await for all I/O operations (database, Redis, external APIs)
- Use `snake_case` for variables and functions (not camelCase)
- Max line length: 100 characters (formatter will enforce)
- Docstrings: Google style, not NumPy style

## Project Structure (Don't Deviate Without Approval)

/app
/api
/v1
/routes # Route definitions only, no business logic
- user_routes.py
- auth_routes.py
/dependencies # FastAPI dependencies (auth, db sessions)
/middleware # Custom middleware
/core
/config.py # Environment config (uses pydantic-settings)
/security.py # Auth helpers, password hashing
/database.py # SQLAlchemy setup
/models # SQLAlchemy ORM models
/schemas # Pydantic request/response schemas
/services # Business logic layer (route → service → repository)
/repositories # Database access layer (one per model)
/tasks # Celery tasks for async jobs
/utils # Helpers, formatters, etc.
/tests
/unit
/integration
/alembic # Database migrations


## Critical Patterns (Read This Before Generating Code)

### 1. Database Sessions - ALWAYS Use Dependency Injection

python

✅ CORRECT - Use this pattern

from app.core.database import get_db
from sqlalchemy.ext.asyncio import AsyncSession

@router.get("/users/{user_id}")
async def get_user(
user_id: int,
db: AsyncSession = Depends(get_db) # DI, not global
):
user = await user_repository.get_by_id(db, user_id)
return user

❌ WRONG - Never do this

global_db = SessionLocal() # NO! This breaks in async context


### 2. Authentication - Secure by Default

python

✅ CORRECT - Routes require auth unless explicitly marked public

@router.get("/profile")
async def get_profile(
current_user: User = Depends(get_current_user) # Auth required
):
return current_user

@router.get("/health")
@public_route # Explicit opt-out decorator
async def health_check():
return {"status": "healthy"}

❌ WRONG - Don't make auth optional by default

async def get_profile(current_user: User | None = Depends(get_current_user_optional))


### 3. Error Handling - Proper HTTP Status Codes

python

✅ CORRECT - Use specific exceptions

from fastapi import HTTPException, status

if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"User {user_id} not found" # Helpful message
)

if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Account is deactivated"
)

❌ WRONG - Generic 400 for everything

raise HTTPException(status_code=400, detail="Error")


### 4. Response Models - Always Specify

python

✅ CORRECT - Explicit response model

@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
user = await user_service.get_user(db, user_id)
return user # Auto-validated and serialized

❌ WRONG - No response model = returns raw SQLAlchemy objects

@router.get("/users/{user_id}") # Missing response_model


### 5. Logging - Use Structured Logging

python

✅ CORRECT - Use logger with context

import logging
from app.core.logging_config import get_logger

logger = get_logger(name)

@router.post("/orders")
async def create_order(order_data: OrderCreate, db: AsyncSession = Depends(get_db)):
logger.info(
"Creating order",
extra={
"user_id": order_data.user_id,
"product_count": len(order_data.items),
"total_amount": order_data.total
}
)
# ... create order ...

❌ WRONG - print() statements

print(f"Order created: {order.id}") # NO! Use logger


### 6. Environment Variables - Type-Safe Config

python

✅ CORRECT - Pydantic settings model

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
DATABASE_URL: str
REDIS_URL: str
SECRET_KEY: str
DEBUG: bool = False
API_RATE_LIMIT: int = 100

model_config = {
    "env_file": ".env",
    "env_file_encoding": "utf-8",
    "case_sensitive": True
}

settings = Settings() # Validates on startup

❌ WRONG - Raw os.getenv()

import os

DB_URL = os.getenv("DATABASE_URL") # No validation, can be None


## Testing Requirements

### Unit Tests

python

✅ CORRECT - Test services in isolation

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_create_user_success():
# Mock the repository layer
mock_repo = AsyncMock()
mock_repo.create.return_value = User(id=1, email="test@example.com")

service = UserService(mock_repo)
result = await service.create_user(UserCreate(email="test@example.com"))

assert result.id == 1
mock_repo.create.assert_called_once()

❌ WRONG - Testing route handlers directly (that's integration testing)


### Integration Tests

python

✅ CORRECT - Test full request/response cycle

from httpx import AsyncClient

@pytest.mark.asyncio
async def test_user_registration_endpoint(client: AsyncClient):
response = await client.post(
"/api/v1/auth/register",
json={"email": "new@example.com", "password": "secure123"}
)

assert response.status_code == 201
assert "id" in response.json()
assert response.json()["email"] == "new@example.com"


## What to NEVER Do (Seriously)

### ❌ Don't Mix Business Logic into Route Handlers

python

WRONG - Route doing too much

@router.post("/orders")
async def create_order(order: OrderCreate, db: AsyncSession = Depends(get_db)):
# Validate inventory
product = await db.execute(...)
if product.stock < order.quantity:
raise HTTPException(400, "Out of stock")

# Calculate pricing
price = product.price * order.quantity
if order.discount_code:
    discount = await db.execute(...)
    price = price * (1 - discount.percentage)

# ... 50 more lines of logic ...

CORRECT - Route delegates to service

@router.post("/orders")
async def create_order(
order: OrderCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
result = await order_service.create_order(db, order, current_user)
return result


### ❌ Don't Use `print()` for Logging

python

WRONG

print(f"User {user_id} logged in")

CORRECT

logger.info("User logged in", extra={"user_id": user_id})


### ❌ Don't Store Secrets in Code

python

WRONG

SECRET_KEY = "hardcoded-secret-key-123"
DATABASE_URL = "postgresql://user:password@localhost/db"

CORRECT - Use environment variables

from app.core.config import settings
SECRET_KEY = settings.SECRET_KEY
DATABASE_URL = settings.DATABASE_URL


### ❌ Don't Skip Input Validation

python

WRONG - Trusting user input

@router.post("/users")
async def create_user(email: str, password: str): # Raw strings, no validation
# ... create user ...

CORRECT - Pydantic validates everything

@router.post("/users", response_model=UserResponse)
async def create_user(user_data: UserCreate): # Pydantic model with validation
# user_data.email is already validated as email format
# user_data.password meets minimum length requirements


### ❌ Don't Return SQLAlchemy Models Directly

python

WRONG - Exposes internal model structure and relationships

@router.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
user = await db.get(User, user_id)
return user # SQLAlchemy model with lazy-loaded relationships

CORRECT - Use Pydantic response schemas

@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
user = await user_repository.get_by_id(db, user_id)
return UserResponse.model_validate(user) # Pydantic v2 syntax


## Performance Rules

1. **Always use connection pooling** - Already configured in core/database.py
2. **Index foreign keys** - Alembic migrations should include indexes
3. **Use select/join loading** - Avoid N+1 queries with SQLAlchemy relationships
4. **Cache frequently accessed data** - Use Redis with 5-minute default TTL
5. **Async everywhere** - If it does I/O, it must be async

## When in Doubt

1. Check existing code in the same module
2. Ask in #backend-squad Slack channel
3. Look at our API documentation: https://docs.internal/api-guidelines
4. Copy patterns from `app/api/v1/routes/user_routes.py` (our reference implementation)

## Last Updated: 2024-01-15
## Questions? Ask @sarah or @mike in #backend-squad

Realistic .cursorrules Template

# .cursorrules Template
# Copy this to your project root and customize for your team

## Project Context
# Project: [Your Project Name]
# Primary Language: [Python/TypeScript/Go/etc]
# Framework: [FastAPI/Express/Django/etc]
# Team Size: [Number]
# Last Updated: [Date]

## Tech Stack
# Be specific with versions - prevents AI suggesting deprecated syntax
- [framework]==[version]
- [database_lib]==[version]
- [orm]==[version]
# Example: fastapi==0.104.1, not just "FastAPI"

## Code Style (Pick One and Stick With It)
# Don't say "follow PEP8" - be specific about YOUR choices
- Indentation: [spaces/tabs] [how many]
- Line length: [80/100/120] characters
- Quotes: [single/double] quotes for strings
- Naming: [camelCase/snake_case/PascalCase] for [what]
- Type hints: [required/optional/forbidden]
- Comments: [where you want them and where you don't]

## Project Structure
# Paste your actual directory tree - AI uses this for imports

/src
/[module_name]
/[subfolder]


## Architecture Patterns

### [Pattern Name] - [When to Use It]

[language]

✅ CORRECT - Show the pattern you WANT

[your preferred code example]

❌ WRONG - Show what you DON'T want (AI learns from negatives)

[anti-pattern you keep seeing in PRs]


### [Another Pattern]

[language]

Include comments explaining WHY, not just WHAT

"We use X because Y keeps breaking in production"


## Testing Requirements
- Test framework: [pytest/jest/etc]
- Coverage requirement: [percentage or "no hard requirement"]
- What must be tested: [list specific things]
- What can skip tests: [scripts, configs, etc]

## Common Mistakes to Avoid
# This section saves hours in PR reviews
- [ ] Don't [common mistake #1]
- [ ] Don't [common mistake #2]
- [ ] Always [thing people forget]

## Dependencies & Environment

bash

How to install everything (so AI can reference correct versions)

[your actual setup commands]


## Gotchas in This Codebase
# The stuff that only exists in YOUR code
- [Legacy pattern you're stuck with]: [explanation]
- [Weird workaround]: [why it exists]
- [That one file nobody touches]: [what it does]

## External Services
# So AI knows what's available
- Database: [type, how to connect]
- Cache: [Redis/Memcached/etc, connection pattern]
- Message Queue: [if applicable]
- Auth: [JWT/OAuth/session, where configured]

## When in Doubt
1. [Where to check existing examples]
2. [Who to ask]
3. [Documentation link]

## Notes
# Anything else that doesn't fit above but matters

Real Example: The Rate Limiting Implementation

Initial Attempt (What Copilot Suggested)

// api/middleware/rate-limit.ts
// Added: 2024-01-10 - Sprint 28 - Security requirement from InfoSec
// Context: Need to prevent API abuse after bot attack last week

import rateLimit from 'express-rate-limit';

// Copilot generated this - seemed reasonable at first
export const apiLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15 minutes
  max: 100, // limit each IP to 100 requests per windowMs
  message: 'Too many requests'
});

// Applied to all routes
app.use('/api/', apiLimiter);

Problems found in code review:

No differentiation between endpoints (login same as health check)
IP-based limiting breaks behind load balancer
No way to allowlist internal services
Generic error message doesn't tell user when to retry

Second Attempt (After Code Review)

// api/middleware/rate-limit.ts
// Updated: 2024-01-12 - Incorporated feedback from code review
// Reviewers: @sarah, @tom
// TODO: Add Redis store once we set up cluster (using memory for now)

import rateLimit, { Options } from 'express-rate-limit';
import { Request, Response } from 'express';

// Get real IP behind load balancer
// Our LB sets X-Forwarded-For header
function getClientIp(req: Request): string {
  const forwardedFor = req.headers['x-forwarded-for'];
  if (typeof forwardedFor === 'string') {
    // First IP in the chain is the original client
    return forwardedFor.split(',')[0].trim();
  }
  return req.ip || req.socket.remoteAddress || 'unknown';
}

// Different limits for different endpoint types
const rateLimitConfigs: Record<string, Partial<Options>> = {
  // Strict limits for auth endpoints (prevent brute force)
  auth: {
    windowMs: 15 * 60 * 1000, // 15 minutes
    max: 5, // 5 attempts per 15 min
    message: {
      error: 'Too many authentication attempts',
      retryAfter: 'Please try again in 15 minutes'
    },
    standardHeaders: true, // Return rate limit info in headers
    legacyHeaders: false,
  },

  // Moderate limits for data modification
  mutation: {
    windowMs: 60 * 1000, // 1 minute
    max: 30, // 30 requests per minute
    message: {
      error: 'Rate limit exceeded',
      retryAfter: 'Please wait a minute before trying again'
    }
  },

  // Relaxed limits for read operations
  query: {
    windowMs: 60 * 1000, // 1 minute  
    max: 100, // 100 requests per minute
    message: {
      error: 'Rate limit exceeded',
      retryAfter: 'Please slow down your requests'
    }
  }
};

// Create rate limiters for each category
export const authLimiter = rateLimit({
  ...rateLimitConfigs.auth,
  keyGenerator: getClientIp,
  // Skip rate limiting for internal services
  skip: (req: Request) => {
    const apiKey = req.headers['x-api-key'];
    return apiKey === process.env.INTERNAL_SERVICE_KEY;
  }
});

export const mutationLimiter = rateLimit({
  ...rateLimitConfigs.mutation,
  keyGenerator: getClientIp,
  skip: (req: Request) => {
    const apiKey = req.headers['x-api-key'];
    return apiKey === process.env.INTERNAL_SERVICE_KEY;
  }
});

export const queryLimiter = rateLimit({
  ...rateLimitConfigs.query,
  keyGenerator: getClientIp,
  skip: (req: Request) => {
    const apiKey = req.headers['x-api-key'];
    return apiKey === process.env.INTERNAL_SERVICE_KEY;
  }
});

// Applied selectively to route groups
// In routes/auth.routes.ts:
// router.post('/login', authLimiter, loginController);
// 
// In routes/users.routes.ts:
// router.get('/users', queryLimiter, getUsersController);
// router.post('/users', mutationLimiter, createUserController);

Still has issues:

Using in-memory store (doesn't work in multi-server setup)
Environment variable check is not secure
No monitoring/alerting when limits are hit

Production Version (Actually Deployed)

// api/middleware/rate-limit.ts
// Version: 3.0 - Production Ready
// Updated: 2024-01-18
// Changes: Added Redis store, proper monitoring, better error handling
// Related: JIRA-1234 (Security audit findings)

import rateLimit, { Options, Store } from 'express-rate-limit';
import RedisStore from 'rate-limit-redis';
import { Request, Response } from 'express';
import { createClient } from 'redis';
import { logger } from '../utils/logger';
import { metricsClient } from '../monitoring/metrics';

// Redis client for rate limit storage
// Shared across all app instances - critical for multi-server setup
const redisClient = createClient({
  url: process.env.REDIS_URL || 'redis://localhost:6379',
  password: process.env.REDIS_PASSWORD,
  socket: {
    reconnectStrategy: (retries) => {
      if (retries > 10) {
        logger.error('Redis connection failed after 10 retries');
        return new Error('Redis connection failed');
      }
      return Math.min(retries * 100, 3000);
    }
  }
});

redisClient.on('error', (err) => {
  logger.error('Redis client error', { error: err.message, stack: err.stack });
  // Alert ops team - rate limiting will fall back to memory
  metricsClient.increment('rate_limit.redis.error');
});

redisClient.on('connect', () => {
  logger.info('Redis client connected for rate limiting');
});

// Initialize Redis connection
(async () => {
  try {
    await redisClient.connect();
  } catch (error) {
    logger.error('Failed to connect to Redis', { error });
    // App continues but rate limiting will be less effective
  }
})();

/**
 * Extract client IP from request, handling proxy scenarios
 * Our infrastructure: Cloudflare -> ALB -> App servers
 */
function getClientIp(req: Request): string {
  // Cloudflare sets CF-Connecting-IP header with original client IP
  const cfIp = req.headers['cf-connecting-ip'];
  if (typeof cfIp === 'string') {
    return cfIp;
  }

  // Fallback to X-Forwarded-For
  const forwardedFor = req.headers['x-forwarded-for'];
  if (typeof forwardedFor === 'string') {
    const ips = forwardedFor.split(',').map(ip => ip.trim());
    // First IP is the original client (before proxies)
    return ips[0];
  }

  // Last resort - direct connection IP
  return req.ip || req.socket.remoteAddress || 'unknown';
}

/**
 * Check if request should bypass rate limiting
 * Used for: internal services, monitoring, health checks
 */
function shouldSkipRateLimit(req: Request): boolean {
  // Health checks from load balancer
  if (req.path === '/health' || req.path === '/metrics') {
    return true;
  }

  // Internal service authentication
  // Uses bcrypt-hashed tokens stored in environment
  const apiKey = req.headers['x-api-key'];
  if (typeof apiKey === 'string') {
    const validKeys = (process.env.INTERNAL_API_KEYS || '').split(',');
    return validKeys.includes(apiKey);
  }

  // Authenticated admin users get higher limits (handled separately)
  // Just skip the base rate limit
  if (req.user?.role === 'admin') {
    return true;
  }

  return false;
}

/**
 * Custom handler when rate limit is exceeded
 * Logs incidents for monitoring and alerts
 */
function rateLimitExceededHandler(
  req: Request,
  res: Response,
  next: () => void,
  options: Options
): void {
  const clientIp = getClientIp(req);

  // Log rate limit violation for security monitoring
  logger.warn('Rate limit exceeded', {
    ip: clientIp,
    path: req.path,
    method: req.method,
    userAgent: req.headers['user-agent'],
    userId: req.user?.id || null,
    limitType: options.windowMs ? `${options.max}/${options.windowMs}ms` : 'unknown'
  });

  // Increment metrics for monitoring dashboard
  metricsClient.increment('rate_limit.exceeded', {
    path: req.path,
    method: req.method
  });

  // Check if this IP is repeatedly hitting limits (possible attack)
  const violationKey = `rate_limit_violations:${clientIp}`;
  redisClient.incr(violationKey).then((violations) => {
    if (violations > 10) {
      logger.error('Possible attack detected - repeated rate limit violations', {
        ip: clientIp,
        violations,
        timeWindow: '1 hour'
      });
      // TODO: Auto-block IPs after threshold (requires approval from security team)
      metricsClient.increment('security.possible_attack_detected');
    }
  }).catch(err => {
    logger.error('Failed to track rate limit violations', { error: err.message });
  });

  // Set violation counter to expire after 1 hour
  redisClient.expire(violationKey, 3600);

  // Return error response with retry information
  res.status(429).json({
    error: 'Too many requests',
    message: options.message || 'You have exceeded the rate limit',
    retryAfter: res.getHeader('Retry-After'),
    // Help developers debug
    limit: options.max,
    windowMs: options.windowMs
  });
}

// Rate limit configurations for different endpoint categories
interface RateLimitConfig extends Partial<Options> {
  name: string;
  description: string;
}

const rateLimitConfigs: Record<string, RateLimitConfig> = {
  // Strictest limits for authentication endpoints
  // Prevents brute force attacks on login
  auth: {
    name: 'auth',
    description: 'Authentication endpoints (login, register, password reset)',
    windowMs: 15 * 60 * 1000, // 15 minutes
    max: 5, // Only 5 attempts per 15 minutes
    message: 'Too many authentication attempts. Please try again later.',
    standardHeaders: true,
    legacyHeaders: false,
    store: new RedisStore({
      client: redisClient,
      prefix: 'rl:auth:',
    }),
  },

  // Moderate limits for data modification
  // Prevents spam and abuse while allowing legitimate use
  mutation: {
    name: 'mutation',
    description: 'Data modification endpoints (POST, PUT, DELETE)',
    windowMs: 60 * 1000, // 1 minute
    max: 30, // 30 mutations per minute
    message: 'Too many requests. Please slow down.',
    standardHeaders: true,
    legacyHeaders: false,
    store: new RedisStore({
      client: redisClient,
      prefix: 'rl:mutation:',
    }),
  },

  // Relaxed limits for read operations
  // Allows dashboards and monitoring to function normally
  query: {
    name: 'query',
    description: 'Read-only endpoints (GET)',
    windowMs: 60 * 1000, // 1 minute
    max: 100, // 100 queries per minute
    message: 'Too many requests. Please slow down.',
    standardHeaders: true,
    legacyHeaders: false,
    store: new RedisStore({
      client: redisClient,
      prefix: 'rl:query:',
    }),
  },

  // Special category for expensive operations
  // Example: PDF generation, bulk exports
  expensive: {
    name: 'expensive',
    description: 'Resource-intensive operations',
    windowMs: 60 * 60 * 1000, // 1 hour
    max: 10, // Only 10 per hour
    message: 'This operation is rate limited. Please try again later.',
    standardHeaders: true,
    legacyHeaders: false,
    store: new RedisStore({
      client: redisClient,
      prefix: 'rl:expensive:',
    }),
  }
};

// Create rate limiter middleware for each category
function createRateLimiter(config: RateLimitConfig) {
  return rateLimit({
    ...config,
    keyGenerator: getClientIp,
    skip: shouldSkipRateLimit,
    handler: rateLimitExceededHandler,
    // Fail open if Redis is down - log error but allow request
    skipFailedRequests: false,
    skipSuccessfulRequests: false,
  });
}

// Export configured rate limiters
export const authLimiter = createRateLimiter(rateLimitConfigs.auth);
export const mutationLimiter = createRateLimiter(rateLimitConfigs.mutation);
export const queryLimiter = createRateLimiter(rateLimitConfigs.query);
export const expensiveLimiter = createRateLimiter(rateLimitConfigs.expensive);

// Cleanup on app shutdown
export async function cleanupRateLimiting(): Promise<void> {
  try {
    await redisClient.quit();
    logger.info('Rate limiting Redis client disconnected');
  } catch (error) {
    logger.error('Error disconnecting rate limiting Redis client', { error });
  }
}

// Usage examples:
// 
// In routes/auth.routes.ts:
// router.post('/login', authLimiter, loginController);
// router.post('/register', authLimiter, registerController);
// router.post('/reset-password', authLimiter, resetPasswordController);
//
// In routes/users.routes.ts:
// router.get('/users', queryLimiter, getUsersController);
// router.post('/users', mutationLimiter, createUserController);
// router.delete('/users/:id', mutationLimiter, deleteUserController);
//
// In routes/reports.routes.ts:
// router.post('/reports/export', expensiveLimiter, exportReportController);

Companion configuration file:

// .env.example
# Add these to your actual .env file

# Redis Configuration (Required for rate limiting in production)
REDIS_URL=redis://localhost:6379
REDIS_PASSWORD=your-secure-password-here

# Internal API Keys (comma-separated, bcrypt hashed)
# Generate with: bcrypt.hash('your-key', 10)
INTERNAL_API_KEYS=$2b$10$...hash1...,$2b$10$...hash2...

# Monitoring (Optional but recommended)
METRICS_ENABLED=true
METRICS_HOST=statsd.internal:8125

This is what production-quality code actually looks like - evolved through multiple iterations, with real comments explaining decisions, proper error handling, monitoring, and all the edge cases you discover when code hits production.

Forem: Antonio Delgado

Production-Ready Code Examples

Production-Ready Code Examples

The Database Query That Still Haunts Me

Version 1: What I Actually Shipped (The Bad One)

Version 2: After Performance Issues Hit Production

.cursorrules for FastAPI Projects (Real Production Version)

✅ CORRECT - Use this pattern

❌ WRONG - Never do this

global_db = SessionLocal() # NO! This breaks in async context

✅ CORRECT - Routes require auth unless explicitly marked public

❌ WRONG - Don't make auth optional by default

async def get_profile(current_user: User | None = Depends(get_current_user_optional))

✅ CORRECT - Use specific exceptions

❌ WRONG - Generic 400 for everything

raise HTTPException(status_code=400, detail="Error")

✅ CORRECT - Explicit response model

❌ WRONG - No response model = returns raw SQLAlchemy objects

@router.get("/users/{user_id}") # Missing response_model

✅ CORRECT - Use logger with context

❌ WRONG - print() statements

print(f"Order created: {order.id}") # NO! Use logger

✅ CORRECT - Pydantic settings model

❌ WRONG - Raw os.getenv()

import os

DB_URL = os.getenv("DATABASE_URL") # No validation, can be None

✅ CORRECT - Test services in isolation

❌ WRONG - Testing route handlers directly (that's integration testing)

✅ CORRECT - Test full request/response cycle

WRONG - Route doing too much

CORRECT - Route delegates to service

WRONG

CORRECT

WRONG

CORRECT - Use environment variables

WRONG - Trusting user input

CORRECT - Pydantic validates everything

WRONG - Exposes internal model structure and relationships

CORRECT - Use Pydantic response schemas

Realistic .cursorrules Template

✅ CORRECT - Show the pattern you WANT

❌ WRONG - Show what you DON'T want (AI learns from negatives)

Include comments explaining WHY, not just WHAT

"We use X because Y keeps breaking in production"

How to install everything (so AI can reference correct versions)

Real Example: The Rate Limiting Implementation

Initial Attempt (What Copilot Suggested)

Second Attempt (After Code Review)

Production Version (Actually Deployed)