# Fastapi

> FastAPI Framework

- Skill: `muhammederem/fastapi` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add muhammederem/fastapi`
- Raw SKILL.md: https://api.skillmd.com/api/skills/muhammederem/fastapi/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: muhammederem (https://skillmd.com/u/muhammederem)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/muhammederem/fastapi

---

# FastAPI Framework

## Overview
FastAPI is a modern, fast web framework for building APIs with Python based on standard Python type hints. It provides automatic validation, serialization, and interactive API documentation.

## Installation
```bash
pip install fastapi uvicorn[standard]
pip install pydantic pydantic-settings
```

## Basic Application

### Hello World
```python
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}
```

### Run Server
```bash
uvicorn main:app --reload
# Or with specific host/port
uvicorn main:app --host 0.0.0.0 --port 8000 --reload
```

## Project Structure

### Recommended Structure
```
myapp/
├── app/
│   ├── __init__.py
│   ├── main.py              # Application entry
│   ├── api/
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/   # API endpoints
│   │   │   └── router.py    # API router
│   ├── core/
│   │   ├── config.py        # Configuration
│   │   ├── security.py      # Security
│   │   └── deps.py          # Dependencies
│   ├── models/              # Database models
│   ├── schemas/             # Pydantic schemas
│   ├── services/            # Business logic
│   └── db/
│       └── database.py      # Database connection
├── tests/
├── alembic.ini
└── requirements.txt
```

### Main Application
```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1.router import api_router

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

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Include routers
app.include_router(api_router, prefix="/api/v1")
```

## Pydantic Schemas

### Request/Response Models
```python
from pydantic import BaseModel, EmailStr, Field
from typing import Optional, List
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[EmailStr] = None

class UserResponse(UserBase):
    id: int
    created_at: datetime
    is_active: bool

    class Config:
        from_attributes = True  # For ORM models
```

### Advanced Schemas
```python
from enum import Enum

class Status(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float = Field(..., gt=0)
    tax: Optional[float] = None
    tags: List[str] = []
    status: Status = Status.INACTIVE

    class Config:
        json_schema_extra = {
            "example": {
                "name": "Item name",
                "price": 10.5,
                "tags": ["electronics", "tech"]
            }
        }
```

## Dependency Injection

### Dependencies
```python
from fastapi import Depends, Header, HTTPException, status

async def get_db():
    """Database session dependency"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

async def get_current_user(
    token: str = Header(...),
    db: Session = Depends(get_db)
):
    """Get authenticated user"""
    user = verify_token(token, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials"
        )
    return user

# Use in endpoint
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user
```

### Class-based Dependencies
```python
class CommonQueryParams:
    def __init__(
        self,
        skip: int = 0,
        limit: int = 100,
        sort: str = "created_at"
    ):
        self.skip = skip
        self.limit = limit
        self.sort = sort

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
    return {"skip": commons.skip, "limit": commons.limit}
```

## Database Integration

### SQLAlchemy Setup
```python
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://user:pass@localhost/db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

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

### Model Definition
```python
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    name = Column(String, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
```

### CRUD Operations
```python
from typing import Generic, TypeVar, Type, List
from pydantic import BaseModel

ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, model: Type[ModelType]):
        self.model = model

    def get(self, db: Session, id: int) -> Optional[ModelType]:
        return db.query(self.model).filter(self.model.id == id).first()

    def get_multi(
        self, db: Session, skip: int = 0, limit: int = 100
    ) -> List[ModelType]:
        return db.query(self.model).offset(skip).limit(limit).all()

    def create(
        self, db: Session, obj_in: CreateSchemaType
    ) -> ModelType:
        obj_in_data = obj_in.dict()
        db_obj = self.model(**obj_in_data)
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def update(
        self,
        db: Session,
        db_obj: ModelType,
        obj_in: UpdateSchemaType
    ) -> ModelType:
        obj_data = obj_in.dict(exclude_unset=True)
        for field in obj_data:
            setattr(db_obj, field, obj_data[field])
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def delete(self, db: Session, id: int) -> ModelType:
        obj = db.query(self.model).get(id)
        db.delete(obj)
        db.commit()
        return obj
```

## Authentication

### JWT Authentication
```python
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

@app.post("/token")
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    user = authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect email or password")
    access_token = create_access_token(data={"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}
```

### Protected Routes
```python
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db)
):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
    except JWTError:
        raise HTTPException(status_code=403, detail="Invalid token")
    user = get_user_by_email(db, email=email)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user
```

## RESTful CRUD Endpoints

### Complete CRUD
```python
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

router = APIRouter(prefix="/items", tags=["items"])

@router.post("/", response_model=ItemResponse)
async def create_item(
    item: ItemCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    """Create item"""
    return crud.item.create(db, item)

@router.get("/", response_model=List[ItemResponse])
async def list_items(
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db)
):
    """List items with pagination"""
    return crud.item.get_multi(db, skip=skip, limit=limit)

@router.get("/{item_id}", response_model=ItemResponse)
async def get_item(
    item_id: int,
    db: Session = Depends(get_db)
):
    """Get single item"""
    item = crud.item.get(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.put("/{item_id}", response_model=ItemResponse)
async def update_item(
    item_id: int,
    item_update: ItemUpdate,
    db: Session = Depends(get_db)
):
    """Update item"""
    item = crud.item.get(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return crud.item.update(db, item, item_update)

@router.delete("/{item_id}")
async def delete_item(
    item_id: int,
    db: Session = Depends(get_db)
):
    """Delete item"""
    item = crud.item.get(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    crud.item.delete(db, item_id)
    return {"message": "Item deleted"}
```

## Advanced Features

### Background Tasks
```python
from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # Send email logic
    pass

@router.post("/signup")
async def signup(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Signup successful"}
```

### WebSockets
```python
from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message received: {data}")
```

### File Upload
```python
from fastapi import UploadFile, File

@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    # Save file
    return {"filename": file.filename}
```

### Exception Handlers
```python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"detail": str(exc)}
    )
```

## Middleware

### Custom Middleware
```python
from fastapi import Request

@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
```

## Testing

### Test Client
```python
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_read_item():
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.json()["id"] == 1
```

## Best Practices

### 1. Project Structure
- Separate concerns (routes, models, services)
- Use dependency injection
- Keep business logic in services

### 2. Validation
- Use Pydantic for all input/output
- Define custom validators
- Provide examples in schemas

### 3. Error Handling
- Use appropriate HTTP status codes
- Provide clear error messages
- Handle exceptions globally

### 4. Performance
- Use async/await
- Optimize database queries
- Cache frequently accessed data
- Use pagination

### 5. Security
- Enable CORS properly
- Use HTTPS in production
- Validate and sanitize inputs
- Implement rate limiting

### 6. Documentation
- Use docstrings for endpoints
- Provide request/response examples
- Document authentication

## Integration

- **PostgreSQL**: SQLAlchemy models
- **MongoDB**: Motor for async
- **Redis**: Caching and sessions
- **Docker**: Containerization
- **AWS Lambda**: Serverless deployment