Auto-Generate FastAPI Endpoints from Schemas in 12 Minutes

# For PostgreSQL
pg_dump --schema-only your_db > schema.sql

# For SQLite
sqlite3 your_db.db .schema > schema.sql

# For MySQL
mysqldump --no-data your_db > schema.sql

pip install fastapi[standard]==1.0.0 anthropic==0.40.0 sqlalchemy==2.0.35 --break-system-packages

# generate_endpoints.py
import anthropic
from pathlib import Path

def generate_fastapi_code(schema_sql: str) -> str:
    """Use Claude to generate FastAPI endpoints from SQL schema."""
    
    client = anthropic.Anthropic()  # Reads ANTHROPIC_API_KEY from env
    
    prompt = f"""Generate production-ready FastAPI 1.0 code from this schema.

Requirements:
- Async route handlers with proper typing
- Pydantic V2 models for request/response
- SQLAlchemy 2.0 async sessions
- HTTP exception handling (404, 422, 500)
- One file per table with CRUD operations
- Include database connection setup

Schema:
{schema_sql}

Return ONLY valid Python code, no explanations."""

    message = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4000,
        messages=[{"role": "user", "content": prompt}]
    )
    
    # Extract code blocks from response
    return message.content[0].text


if __name__ == "__main__":
    schema = Path("schema.sql").read_text()
    
    # Generate code (uses prompt caching on repeated runs)
    fastapi_code = generate_fastapi_code(schema)
    
    # Save to file
    Path("generated_api.py").write_text(fastapi_code)
    print("✓ Generated FastAPI code in generated_api.py")

# validate.py
import ast
import subprocess
from pathlib import Path

def validate_generated_code(filepath: str) -> bool:
    """Check if generated code is syntactically valid and type-safe."""
    
    code = Path(filepath).read_text()
    
    # 1. Check Python syntax
    try:
        ast.parse(code)
    except SyntaxError as e:
        print(f"✗ Syntax error: {e}")
        return False
    
    # 2. Check types with mypy
    result = subprocess.run(
        ["mypy", filepath, "--strict"],
        capture_output=True,
        text=True
    )
    
    if result.returncode != 0:
        print(f"✗ Type errors:\n{result.stdout}")
        return False
    
    # 3. Try importing (checks dependencies)
    try:
        subprocess.run(
            ["python", "-c", f"import sys; sys.path.insert(0, '.'); import {Path(filepath).stem}"],
            check=True,
            capture_output=True
        )
    except subprocess.CalledProcessError as e:
        print(f"✗ Import failed: {e.stderr.decode()}")
        return False
    
    print("✓ Code is valid and type-safe")
    return True


if __name__ == "__main__":
    validate_generated_code("generated_api.py")

python validate.py

# Before deploying, add custom validation
# Insert this into generated routes:

from fastapi import HTTPException

@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate, db: AsyncSession):
    # AI generates basic CRUD, you add business rules
    if await db.scalar(select(User).where(User.email == user.email)):
        raise HTTPException(409, "Email already exists")
    
    # AI-generated code continues here...
    new_user = User(**user.model_dump())
    db.add(new_user)
    await db.commit()
    return new_user

# Start the server
uvicorn generated_api:app --reload

# Test an endpoint
curl http://localhost:8000/users/1

pip install locust --break-system-packages
locust -f load_test.py --headless -u 100 -r 10 --run-time 30s

# Generated from: CREATE TABLE users (id SERIAL PRIMARY KEY, email VARCHAR(255) UNIQUE, created_at TIMESTAMP)

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from sqlalchemy import Column, Integer, String, DateTime, select
from datetime import datetime

# Database setup
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

# Pydantic models
class UserBase(BaseModel):
    email: EmailStr

class UserCreate(UserBase):
    pass

class UserResponse(UserBase):
    id: int
    created_at: datetime
    
    class Config:
        from_attributes = True

# FastAPI app
app = FastAPI(title="Generated User API")

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    
    if not user:
        raise HTTPException(404, f"User {user_id} not found")
    
    return user

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    new_user = User(email=user.email)
    db.add(new_user)
    
    try:
        await db.commit()
        await db.refresh(new_user)
    except Exception:
        await db.rollback()
        raise HTTPException(500, "Failed to create user")
    
    return new_user