FastAPI

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List

app = FastAPI(title="My API", version="1.0")

# Mock DB
fake_db = []
id_counter = 1

class UserCreate(BaseModel):
    name: str
    email: str
    age: int = 18

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        from_attributes = True  # use orm_mode=True if Pydantic v1

@app.get("/users", response_model=List[UserResponse])
async def list_users():
    return fake_db

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

@app.post("/users", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    global id_counter
    new_user = {
        "id": id_counter,
        "name": user.name,
        "email": user.email,
    }
    fake_db.append(new_user)
    id_counter += 1
    return new_user

from fastapi import FastAPI

app = FastAPI(title="Production API")

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

# Rule-of-thumb matrix (pseudo-code as comments)
# If you need:
# - API-first + async external calls + OpenAPI docs -> FastAPI
# - Very small sync service or prototype -> Flask
# - Full web app + admin + ORM + templates -> Django

USE_FASTAPI_WHEN = [
    "many API endpoints",
    "strict input/output schemas",
    "high throughput with async I/O",
    "multiple client teams relying on API docs",
]

AVOID_FASTAPI_WHEN = [
    "you need Django admin-like productivity out of the box",
    "team lacks async basics and cannot invest now",
    "your app is mostly server-rendered pages",
]

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = await verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

async def require_admin(current_user = Depends(get_current_user)):
    if not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Forbidden")
    return current_user

# Use in routes
@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    _: User = Depends(require_admin),  # Auth enforced
):
    await db.delete_user(user_id)
    return {"message": "Deleted"}

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

pip install --upgrade pip
pip install fastapi uvicorn[standard] pydantic-settings sqlalchemy alembic python-jose[cryptography] passlib[bcrypt]

# optional dev tools
pip install pytest httpx ruff mypy

from fastapi import FastAPI

app = FastAPI()

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

# Run:
# uvicorn app.main:app --reload

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()

class TodoCreate(BaseModel):
    title: str = Field(min_length=3, max_length=120)
    done: bool = False

class TodoOut(BaseModel):
    id: int
    title: str
    done: bool

TODOS = [{"id": 1, "title": "Ship API", "done": False}]

@app.get("/todos/{todo_id}", response_model=TodoOut)
def get_todo(todo_id: int):
    return next(t for t in TODOS if t["id"] == todo_id)

@app.get("/todos", response_model=list[TodoOut])
def list_todos(limit: int = Query(default=10, ge=1, le=100)):
    return TODOS[:limit]

@app.post("/todos", response_model=TodoOut, status_code=201)
def create_todo(payload: TodoCreate):
    todo = {"id": len(TODOS) + 1, **payload.model_dump()}
    TODOS.append(todo)
    return todo

app/
  main.py
  api/
    v1/
      routes_todos.py
  schemas/
    todo.py
  services/
    todo_service.py
  repositories/
    todo_repo.py
  db/
    session.py
    models.py
  core/
    config.py
    security.py

# app/api/v1/routes_todos.py
from fastapi import APIRouter, Depends
from app.schemas.todo import TodoCreate, TodoOut
from app.services.todo_service import TodoService

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

@router.post("", response_model=TodoOut)
def create_todo(payload: TodoCreate, service: TodoService = Depends()):
    return service.create(payload)

# app/services/todo_service.py
class TodoService:
    def create(self, payload):
        # business rules go here (dedupe, audit, etc.)
        return {"id": 123, **payload.model_dump()}

from datetime import datetime, timedelta, timezone
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError

SECRET = "change-me"
ALGO = "HS256"
security = HTTPBearer()
app = FastAPI()

@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Service"] = "fastapi-todo"
    return response

def create_access_token(sub: str, minutes: int = 30):
    exp = datetime.now(timezone.utc) + timedelta(minutes=minutes)
    return jwt.encode({"sub": sub, "exp": exp}, SECRET, algorithm=ALGO)

def get_current_user(creds: HTTPAuthorizationCredentials = Depends(security)):
    token = creds.credentials
    try:
        payload = jwt.decode(token, SECRET, algorithms=[ALGO])
        return payload["sub"]
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/me")
def me(user_id: str = Depends(get_current_user)):
    return {"user_id": user_id}

from sqlalchemy import create_engine, String, Boolean
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker, Session
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

class Base(DeclarativeBase):
    pass

class Todo(Base):
    __tablename__ = "todos"
    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(120), index=True)
    done: Mapped[bool] = mapped_column(Boolean, default=False)

Base.metadata.create_all(bind=engine)

class TodoCreate(BaseModel):
    title: str

class TodoOut(BaseModel):
    id: int
    title: str
    done: bool

app = FastAPI()

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

@app.post("/db/todos", response_model=TodoOut)
def create_todo(payload: TodoCreate, db: Session = Depends(get_db)):
    todo = Todo(title=payload.title)
    db.add(todo)
    db.commit()
    db.refresh(todo)
    return todo

@app.get("/db/todos/{todo_id}", response_model=TodoOut)
def get_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = db.get(Todo, todo_id)
    if not todo:
        raise HTTPException(404, "Todo not found")
    return todo

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic_settings import BaseSettings
import logging

class Settings(BaseSettings):
    app_name: str = "todo-api"
    debug: bool = False
    jwt_secret: str
    class Config:
        env_file = ".env"

settings = Settings()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("api")

app = FastAPI(title=settings.app_name)

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

@app.get("/boom")
def boom():
    logger.info("Boom endpoint called")
    raise ValueError("Invalid business input")

# Dockerfile (minimal)
# FROM python:3.12-slim
# WORKDIR /app
# COPY requirements.txt .
# RUN pip install -r requirements.txt
# COPY . .
# CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

from fastapi import FastAPI
import httpx
from fastapi.concurrency import run_in_threadpool

app = FastAPI()

# Good async I/O
@app.get("/external")
async def call_external():
    async with httpx.AsyncClient(timeout=5.0) as client:
        r = await client.get("https://httpbin.org/get")
    return {"status": r.status_code}

# If you must call blocking code, move it off event loop

def blocking_cpu_or_legacy_call():
    return sum(range(10_000_000))

@app.get("/blocking-safe")
async def blocking_safe():
    result = await run_in_threadpool(blocking_cpu_or_legacy_call)
    return {"result": result}

# Example curl:
# curl -H "Authorization: Bearer <token>" http://localhost:8000/me