I'm building a REST API for [describe your domain]. The key entities are:
- [Entity 1]: [attributes]
- [Entity 2]: [attributes]

Generate Pydantic v2 models for these entities, including:
- Base, Create, Update, and Response variants
- Appropriate field validators (email, UUID, constrained strings)
- Relationships between models
- Config for ORM mode where relevant

pythonfrom pydantic import BaseModel, EmailStr, Field, ConfigDict
from typing import Optional
from datetime import datetime
from uuid import UUID, uuid4
from enum import Enum

class TaskStatus(str, Enum):
    pending = "pending"
    in_progress = "in_progress"
    done = "done"
    cancelled = "cancelled"

class TaskPriority(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"
    critical = "critical"

# Base model shared by all task variants
class TaskBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = Field(None, max_length=2000)
    priority: TaskPriority = TaskPriority.medium
    due_date: Optional[datetime] = None

class TaskCreate(TaskBase):
    assignee_email: Optional[EmailStr] = None

class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = None
    status: Optional[TaskStatus] = None
    priority: Optional[TaskPriority] = None
    due_date: Optional[datetime] = None

class TaskResponse(TaskBase):
    model_config = ConfigDict(from_attributes=True)
    
    id: UUID = Field(default_factory=uuid4)
    status: TaskStatus = TaskStatus.pending
    created_at: datetime
    updated_at: datetime
    assignee_email: Optional[EmailStr] = None

Using these Pydantic models, write a FastAPI router for Task CRUD operations.
Database: SQLAlchemy 2.0 async with PostgreSQL.
Requirements:
- Async route handlers throughout
- Dependency injection for the database session
- Proper HTTP status codes (201 for creates, 204 for deletes)
- Pagination on list endpoints (limit/offset)
- OpenAPI docstrings on each route

pythonfrom fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from uuid import UUID
from typing import List

from .database import get_db
from .models import Task
from .schemas import TaskCreate, TaskUpdate, TaskResponse

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

@router.get("/", response_model=List[TaskResponse])
async def list_tasks(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    status: Optional[TaskStatus] = None,
    db: AsyncSession = Depends(get_db),
):
    """
    Retrieve paginated list of tasks.
    
    - **skip**: number of records to skip (for pagination)
    - **limit**: max records returned (1-100, default 20)
    - **status**: optional filter by task status
    """
    query = select(Task)
    if status:
        query = query.where(Task.status == status)
    query = query.offset(skip).limit(limit)
    result = await db.execute(query)
    return result.scalars().all()

@router.post("/", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(
    task_in: TaskCreate,
    db: AsyncSession = Depends(get_db),
):
    """Create a new task."""
    task = Task(**task_in.model_dump())
    db.add(task)
    await db.commit()
    await db.refresh(task)
    return task

@router.get("/{task_id}", response_model=TaskResponse)
async def get_task(
    task_id: UUID,
    db: AsyncSession = Depends(get_db),
):
    """Retrieve a single task by ID."""
    result = await db.execute(select(Task).where(Task.id == task_id))
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@router.patch("/{task_id}", response_model=TaskResponse)
async def update_task(
    task_id: UUID,
    task_in: TaskUpdate,
    db: AsyncSession = Depends(get_db),
):
    """Partially update a task. Only provided fields are updated."""
    result = await db.execute(select(Task).where(Task.id == task_id))
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    
    update_data = task_in.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(task, field, value)
    
    await db.commit()
    await db.refresh(task)
    return task

@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(
    task_id: UUID,
    db: AsyncSession = Depends(get_db),
):
    """Permanently delete a task."""
    result = await db.execute(select(Task).where(Task.id == task_id))
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    await db.delete(task)
    await db.commit()

Add JWT authentication to this FastAPI app:
- POST /auth/register — create user, hash password with bcrypt
- POST /auth/token — return access token (HS256, 30 min expiry)
- Reusable get_current_user dependency that reads Bearer token
- Protect all task routes behind this dependency
Use: python-jose, passlib[bcrypt], pydantic-settings for config

Write pytest tests for the task CRUD endpoints using:
- pytest-asyncio and httpx AsyncClient for async test execution
- An in-memory SQLite database (not the production DB)
- Fixtures for app client, authenticated user token, and sample task
- Test all happy paths and key error cases (404, 422 validation errors)

pythonimport pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.database import get_db
from app.models import Base

TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest.fixture
async def db_session():
    engine = create_async_engine(TEST_DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with async_session() as session:
        yield session
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session
    
    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

@pytest.mark.asyncio
async def test_create_task(client, auth_headers):
    response = await client.post(
        "/tasks/",
        json={"title": "Write tests", "priority": "high"},
        headers=auth_headers,
    )
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Write tests"
    assert data["status"] == "pending"
    assert "id" in data

@pytest.mark.asyncio
async def test_create_task_missing_title(client, auth_headers):
    response = await client.post(
        "/tasks/",
        json={"priority": "low"},
        headers=auth_headers,
    )
    assert response.status_code == 422  # Pydantic validation error

@pytest.mark.asyncio
async def test_get_task_not_found(client, auth_headers):
    import uuid
    response = await client.get(f"/tasks/{uuid.uuid4()}", headers=auth_headers)
    assert response.status_code == 404

bash# Inside your project directory with Claude Code
claude "Add a background task endpoint that sends an email notification when a task 
is marked as done. Use FastAPI BackgroundTasks. Mock the email send for now 
but structure it so a real SMTP integration is a one-line change."

Review this FastAPI app for:
1. Any synchronous I/O calls inside async route handlers
2. Missing input validation on query parameters
3. Routes that don't handle database errors with proper HTTP status codes
4. Any endpoint that could expose internal error messages to the client
5. Missing rate limiting on auth endpoints

Generate a docker-compose.yml for local development of this FastAPI app with:
- App service (reload enabled)
- PostgreSQL 16
- Redis for caching
- Health checks on all services
- Named volumes for data persistence