Building REST APIs with FastAPI

# pip install fastapi uvicorn[standard] pydantic
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import Optional, List
import uvicorn

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

# GET endpoint
@app.get("/")
async def root():
    return {"message": "Hello, World!"}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

# Run the server
if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

# Access at: http://localhost:8000
# Interactive docs: http://localhost:8000/docs
# Alternative docs: http://localhost:8000/redoc

@app.get("/users/{user_id}")
async def get_user(user_id: int):  # Type hint validates and converts
    # FastAPI returns 422 if user_id isn't a valid integer
    return {"user_id": user_id}

@app.get("/items/{item_id}")
async def get_item(
    item_id: int,
    q: Optional[str] = None,          # Optional query param
    limit: int = 10,                   # Default value
    offset: int = 0
):
    result = {"item_id": item_id, "limit": limit, "offset": offset}
    if q:
        result["query"] = q
    return result

from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional
from datetime import datetime

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, pattern="^[a-zA-Z0-9_]+$")
    email: EmailStr  # pip install email-validator
    full_name: str = Field(..., min_length=2)
    age: Optional[int] = Field(None, ge=0, le=150)
    
    @validator('username')
    def username_lowercase(cls, v):
        return v.lower()

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    full_name: str
    created_at: datetime
    
    class Config:
        from_attributes = True  # Allow creating from ORM objects

# In-memory database (in production, use a real DB)
fake_db: dict[int, dict] = {}
next_id = 1

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    global next_id
    new_user = {
        "id": next_id,
        **user.model_dump(),
        "created_at": datetime.utcnow()
    }
    fake_db[next_id] = new_user
    next_id += 1
    return new_user

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    if user_id not in fake_db:
        raise HTTPException(status_code=404, detail=f"User {user_id} not found")
    return fake_db[user_id]

@app.get("/users", response_model=List[UserResponse])
async def list_users(skip: int = 0, limit: int = 10):
    users = list(fake_db.values())
    return users[skip : skip + limit]

@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: int):
    if user_id not in fake_db:
        raise HTTPException(status_code=404, detail="User not found")
    del fake_db[user_id]

from fastapi import Depends
from functools import lru_cache

# Configuration dependency
class Settings:
    app_name: str = "My App"
    database_url: str = "sqlite:///./app.db"
    api_key: str = "secret"

@lru_cache
def get_settings() -> Settings:
    return Settings()

# Authentication dependency
async def get_current_user(
    token: str,
    settings: Settings = Depends(get_settings)
):
    if token != settings.api_key:
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"}
        )
    return {"username": "alice"}

# Use dependencies in endpoints
@app.get("/protected")
async def protected_route(current_user: dict = Depends(get_current_user)):
    return {"message": f"Hello, {current_user['username']}!"}

from fastapi import BackgroundTasks
import smtplib
from email.message import EmailMessage

def send_welcome_email(email: str, username: str):
    """Runs in background — doesn't block the response."""
    # Simulate sending email
    print(f"Sending welcome email to {email}...")
    import time
    time.sleep(2)
    print(f"Email sent to {email}")

@app.post("/register")
async def register(user: UserCreate, background_tasks: BackgroundTasks):
    # Create user immediately
    new_user = create_user_in_db(user)
    
    # Send email in background — response returns immediately
    background_tasks.add_task(
        send_welcome_email, 
        email=user.email, 
        username=user.username
    )
    
    return {"message": "User created", "user_id": new_user["id"]}

from fastapi.middleware.cors import CORSMiddleware
import time
import logging

logger = logging.getLogger(__name__)

# CORS — allow frontend to call your API
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Request logging middleware
@app.middleware("http")
async def log_requests(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    logger.info(
        f"{request.method} {request.url.path} "
        f"→ {response.status_code} in {duration:.3f}s"
    )
    return response

from fastapi import Request
from fastapi.responses import JSONResponse

class DatabaseError(Exception):
    pass

@app.exception_handler(DatabaseError)
async def database_error_handler(request: Request, exc: DatabaseError):
    return JSONResponse(
        status_code=503,
        content={"detail": "Database unavailable", "error": str(exc)}
    )

@app.exception_handler(404)
async def not_found_handler(request: Request, exc):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Route {request.url.path} not found"}
    )

# routers/users.py
from fastapi import APIRouter

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

@router.get("/")
async def list_users():
    return []

@router.post("/")
async def create_user():
    return {}

# main.py
from routers import users, products, orders

app.include_router(users.router)
app.include_router(products.router, prefix="/v1")
app.include_router(orders.router)