FastAPI — SKILL.md

---
name: "FastAPI"
description: "Skill for FastAPI — auto-generated from documentation"
version: "1.0.0"
author: "skynet"
category: "dev"
agents: ["claude-code", "codex", "gemini"]
tags: ["fastapi", "dev", "auto-generated"]
---

# FastAPI

---
name: FastAPI
description: Use when building high-performance Python web APIs with automatic documentation, type hints, and async support. Essential for REST APIs, microservices, and data validation.
category: dev
metadata:
  author: skynet
  version: 1.0.0
---

# FastAPI

## Installation

```bash
# Basic installation
pip install fastapi uvicorn

# With all optional dependencies
pip install "fastapi[all]"

# For production
pip install fastapi uvicorn[standard] gunicorn
```

## Quick Start

```python
# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}
```

```bash
# Development server
uvicorn main:app --reload

# Production server
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
```

## Pydantic Models & Validation

```python
from pydantic import BaseModel, EmailStr, validator
from datetime import datetime
from typing import Optional, List

class UserBase(BaseModel):
    email: EmailStr
    name: str
    age: Optional[int] = None

class UserCreate(UserBase):
    password: str
    
    @validator('password')
    def validate_password(cls, v):
        if len(v) < 8:
            raise ValueError('Password must be at least 8 characters')
        return v

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

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate):
    # Process user creation
    return user
```

## Path & Query Parameters

```python
from enum import Enum
from typing import Optional, List

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
    return {"model_name": model_name}

@app.get("/items/")
def read_items(
    skip: int = 0,
    limit: int = 100,
    tags: List[str] = Query(None, description="Tags to filter by"),
    q: Optional[str] = Query(None, min_length=3, max_length=50)
):
    return {"skip": skip, "limit": limit, "tags": tags, "q": q}
```

## Request Body & Forms

```python
from fastapi import Form, File, UploadFile

# JSON body
@app.post("/items/")
def create_item(item: Item):
    return item

# Form data
@app.post("/login/")
def login(username: str = Form(), password: str = Form()):
    return {"username": username}

# File upload
@app.post("/files/")
def upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename, "size": file.size}

# Multiple files
@app.post("/uploadfiles/")
def upload_files(files: List[UploadFile] = File(...)):
    return [{"filename": f.filename} for f in files]
```

## Dependency Injection

```python
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db)
):
    token = credentials.credentials
    user = verify_token(token, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials"
        )
    return user

@app.get("/protected/")
def protected_route(current_user: User = Depends(get_current_user)):
    return {"user": current_user}
```

## Middleware & CORS

```python
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import time

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

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

## Error Handling

```python
from fastapi import HTTPException
from fastapi.responses import JSONResponse

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something wrong."}
    )

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id == 0:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "Custom error header"}
        )
    return {"item_id": item_id}
```

## Async Operations

```python
import asyncio
import httpx

@app.get("/async-data")
async def get_async_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

@app.post("/process-async/")
async def process_async(data: ProcessData):
    # Simulate async processing
    await asyncio.sleep(1)
    result = await process_data_async(data)
    return {"result": result, "processed_at": datetime.utcnow()}
```

## Database Integration (SQLAlchemy)

```python
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# Database setup
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Models
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    name = Column(String)

# Database operations
@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user
```

## Testing

```python
# test_main.py
from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

def test_create_user():
    user_data = {
        "email": "test@example.com",
        "name": "Test User",
        "password": "testpassword123"
    }
    response = client.post("/users/", json=user_data)
    assert response.status_code == 200
    assert response.json()["email"] == user_data["email"]

# Run tests
pytest test_main.py -v
```

## Decision Tree

```
Choose FastAPI approach:
├── Simple CRUD API? → Use Pydantic models + SQLAlchemy
├── High-performance async? → Use async/await with asyncio
├── File uploads? → Use UploadFile with Form data
├── Authentication needed?
│   ├── Simple token? → Use HTTPBearer dependency
│   └── OAuth2/JWT? → Use FastAPI security utilities
├── External API calls? → Use httpx with async client
└── Real-time features? → Consider WebSocket support
```

## Deployment Commands

```bash
# Docker
echo "FROM python:3.9
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD uvicorn main:app --host 0.0.0.0 --port 8000" > Dockerfile

docker build -t my-api .
docker run -p 8000:8000 my-api

# Gunicorn for production
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

# Environment variables
export DATABASE_URL="postgresql://user:pass@localhost/db"
export SECRET_KEY="your-secret-key"
uvicorn main:app --env-file .env
```

## Troubleshooting

**Error: `422 Unprocessable Entity`**
```
Fix: Check Pydantic model validation
- Verify required fields are present
- Check data types match model definition
- Add proper validators for custom validation
```

**Error: `ImportError: No module named 'email_validator'`**
```bash
pip install email-validator
```

**Error: `RuntimeError: cannot be called from a running async loop`**
```python
# Wrong
response = requests.get(url)

# Right
async with httpx.AsyncClient() as client:
    response = await client.get(url)
```

**Error: `sqlalchemy.exc.OperationalError: connection failed`**
```python
# Check database URL and ensure DB is running
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost:5432/dbname"

# Test connection
from sqlalchemy import create_engine
engine = create_engine(SQLALCHEMY_DATABASE_URL)
engine.connect()
```