FastAPI mit Claude Code: Moderne Python APIs 2026

# Prompt an Claude Code:
"Erstelle ein FastAPI-Projekt mit Auth, Users und Items Router.
Nutze Pydantic v2, python-jose für JWT, passlib für Hashing."

# Claude Code generiert und installiert:
pip install fastapi uvicorn[standard] pydantic[email]
pip install python-jose[cryptography] passlib[bcrypt]
pip install sqlalchemy alembic httpx pytest

myapi/
├── app/
│   ├── main.py            # FastAPI App + Router-Registrierung
│   ├── core/
│   │   ├── config.py      # Settings via pydantic-settings
│   │   ├── security.py    # JWT + Password Hashing
│   │   └── deps.py        # Dependency Injection
│   ├── api/
│   │   ├── v1/
│   │   │   ├── auth.py
│   │   │   ├── users.py
│   │   │   └── items.py
│   ├── models/            # SQLAlchemy ORM Models
│   ├── schemas/           # Pydantic Request/Response Schemas
│   └── services/          # Business Logic
├── tests/
├── Dockerfile
└── docker-compose.yml

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1 import auth, users, items
from app.core.config import settings

app = FastAPI(
    title=settings.PROJECT_NAME,
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.BACKEND_CORS_ORIGINS,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router, prefix="/api/v1/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
app.include_router(items.router, prefix="/api/v1/items", tags=["items"])

@app.get("/")
async def root():
    return {"message": "API v1 ready", "docs": "/docs"}

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional, Annotated
from datetime import datetime
import uuid

class ItemBase(BaseModel):
    title: Annotated[str, Field(min_length=1, max_length=200)]
    description: Optional[str] = None
    price: Annotated[float, Field(gt=0, le=99999)]
    is_active: bool = True

    @field_validator("title")
    @classmethod
    def title_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Titel darf nicht leer sein")
        return v.strip()

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    is_active: Optional[bool] = None

class ItemResponse(ItemBase):
    id: uuid.UUID
    owner_id: uuid.UUID
    created_at: datetime
    updated_at: Optional[datetime] = None

    class Config:
        from_attributes = True  # Pydantic v2: ORM mode

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.schemas.item import ItemCreate, ItemUpdate, ItemResponse
from app.core.deps import get_db, get_current_user

router = APIRouter()

@router.get("/", response_model=list[ItemResponse])
async def list_items(
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=20, le=100),
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user),
):
    return await item_service.get_multi(db, owner_id=current_user.id, skip=skip, limit=limit)

@router.post("/", response_model=ItemResponse, status_code=201)
async def create_item(
    item_in: ItemCreate,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user),
):
    return await item_service.create(db, item_in=item_in, owner_id=current_user.id)

@router.get("/{item_id}", response_model=ItemResponse)
async def get_item(item_id: uuid.UUID, db = Depends(get_db)):
    item = await item_service.get(db, id=item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item nicht gefunden")
    return item

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import AsyncSessionLocal
from app.core.security import verify_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_db() -> AsyncSession:
    """Async DB Session per Request — auto-close via context manager."""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
):
    user_id = verify_token(token)
    if not user_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token ungültig oder abgelaufen",
            headers={"WWW-Authenticate": "Bearer"},
        )
    user = await user_service.get(db, id=user_id)
    if not user or not user.is_active:
        raise HTTPException(status_code=403, detail="Account deaktiviert")
    return user

async def get_current_superuser(current_user = Depends(get_current_user)):
    if not current_user.is_superuser:
        raise HTTPException(status_code=403, detail="Superuser-Rechte erforderlich")
    return current_user

from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings

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

def create_access_token(subject: str, expires_delta: timedelta = None) -> str:
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=30))
    to_encode = {"exp": expire, "sub": str(subject), "type": "access"}
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

def verify_token(token: str) -> str | None:
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        return payload.get("sub")
    except JWTError:
        return None

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from app.schemas.token import Token
from app.core.security import verify_password, create_access_token

router = APIRouter()

@router.post("/login", response_model=Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db = Depends(get_db),
):
    user = await user_service.get_by_email(db, email=form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Ungültige Zugangsdaten")
    token = create_access_token(subject=str(user.id))
    return {"access_token": token, "token_type": "bearer"}

from fastapi import BackgroundTasks
import asyncio, httpx

async def send_notification(user_email: str, item_title: str):
    """Läuft nach dem Response — blockiert Client nicht."""
    async with httpx.AsyncClient() as client:
        await client.post(
            "https://api.sendgrid.com/v3/mail/send",
            json={"to": user_email, "subject": f"Item '{item_title}' erstellt"},
        )

@router.post("/items/", response_model=ItemResponse, status_code=201)
async def create_item_with_notification(
    item_in: ItemCreate,
    background_tasks: BackgroundTasks,
    db = Depends(get_db),
    current_user = Depends(get_current_user),
):
    item = await item_service.create(db, item_in=item_in, owner_id=current_user.id)
    # Response sofort zurück, E-Mail im Hintergrund
    background_tasks.add_task(send_notification, current_user.email, item.title)
    return item

# Parallel async calls mit asyncio.gather()
@router.get("/dashboard")
async def dashboard(db = Depends(get_db), user = Depends(get_current_user)):
    items, stats, recent = await asyncio.gather(
        item_service.get_user_items(db, user.id),
        stats_service.get_summary(db, user.id),
        activity_service.get_recent(db, user.id),
    )
    return {"items": items, "stats": stats, "recent": recent}

@router.post(
    "/items/",
    response_model=ItemResponse,
    status_code=201,
    summary="Neues Item erstellen",
    description="""
    Erstellt ein neues Item für den authentifizierten User.

    - **title**: Pflichtfeld, 1-200 Zeichen
    - **price**: Muss positiv sein (> 0)
    - **is_active**: Standardmäßig True
    """,
    responses={
        201: {"description": "Item erfolgreich erstellt"},
        422: {"description": "Validierungsfehler"},
        401: {"description": "Nicht authentifiziert"},
    },
    tags=["items"],
)
async def create_item(item_in: ItemCreate, ...):
    ...

FROM python:3.12-slim

WORKDIR /app

# System-Deps (nur was nötig ist)
RUN apt-get update && apt-get install -y --no-install-recommends \
    libpq-dev gcc && rm -rf /var/lib/apt/lists/*

# Dependencies zuerst (Layer-Cache nutzen)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY ./app ./app

# Non-root User für Security
RUN useradd -m appuser && chown -R appuser /app
USER appuser

EXPOSE 8000

CMD ["uvicorn", "app.main:app",
     "--host", "0.0.0.0",
     "--port", "8000",
     "--workers", "4",
     "--log-level", "info"]

services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      DATABASE_URL: postgresql+asyncpg://user:pass@db:5432/myapi
      SECRET_KEY: ${SECRET_KEY}
    depends_on:
      db:
        condition: service_healthy
    restart: unless-stopped

  db:
    image: postgres:16-alpine
    volumes: [pgdata:/var/lib/postgresql/data]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d myapi"]
      interval: 5s
      retries: 5

volumes:
  pgdata: