Module 3: Advanced Response Models

Project: Professional Responses for the Task Manager API

Next →

Project overview

In this project you bring together everything you learned in Module 3: schemas differentiated by context, CSV export with streaming, custom headers, cookies, and complete response documentation in OpenAPI. Your Task Manager API goes from "returns everything as JSON" to "every endpoint returns exactly what the consumer needs, in the format they need, with the documentation they need."

The result is an API that serves three kinds of consumers: a web frontend that needs compact data with no internal fields, an admin dashboard that needs everything, and a reporting system that needs CSV streaming for large datasets. One codebase, three different experiences. That's professional response control.

Project objectives

By the time you finish this project:

  • ✅ Three response schemas per resource: TaskSummary, TaskPublic, TaskAdmin
  • ✅ A CSV export endpoint with StreamingResponse and a generator
  • ✅ Custom headers: X-Total-Count, X-Export-Count, X-API-Version
  • ✅ response_model_exclude_unset on the PATCH endpoint
  • ✅ Consistent error responses with an ErrorResponse schema
  • ✅ Complete OpenAPI documentation with the responses parameter on every endpoint
  • ✅ A reusable response wrapper with standard headers
  • ✅ A stats endpoint with an aggregated view

Project structure

fastapi-advanced/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py          ← Pydantic schemas (Task*, ErrorResponse)
│   ├── data.py             ← In-memory data
│   ├── responses.py        ← Response helpers and wrappers
│   └── routers/
│       ├── __init__.py
│       └── tasks.py        ← The tasks router with professional responses
└── requirements.txt

Step 1: Pydantic schemas — app/models.py

Define every schema you'll use. The hierarchy keeps public fields separate from internal ones:

from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
from typing import Optional, Literal


class TaskBase(BaseModel):
    """Editable fields common to create and update."""
    title: str = Field(min_length=1, max_length=200)
    description: Optional[str] = Field(default=None, max_length=2000)
    status: Literal["pending", "in_progress", "completed"] = "pending"
    priority: int = Field(default=1, ge=1, le=5)
    due_date: Optional[str] = None


class TaskCreate(TaskBase):
    """Schema for creating a task (POST)."""
    pass


class TaskUpdate(BaseModel):
    """Schema for partial updates (PATCH). Everything is optional."""
    title: Optional[str] = Field(default=None, min_length=1, max_length=200)
    description: Optional[str] = Field(default=None, max_length=2000)
    status: Optional[Literal["pending", "in_progress", "completed"]] = None
    priority: Optional[int] = Field(default=None, ge=1, le=5)
    due_date: Optional[str] = None


class TaskSummary(BaseModel):
    """Compact view for listings."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    status: str
    priority: int


class TaskPublic(TaskBase):
    """Public view — no internal fields."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    created_at: datetime


class TaskAdmin(TaskBase):
    """Admin view — every field visible."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    created_at: datetime
    updated_at: Optional[datetime] = None
    created_by: str
    internal_notes: Optional[str] = None
    is_archived: bool


class TaskInDB(TaskAdmin):
    """The complete model in storage."""
    pass


class ErrorResponse(BaseModel):
    """The standard schema for error responses."""
    status: Literal["error"] = "error"
    message: str
    detail: Optional[str] = None
    error_code: str


class PaginationMeta(BaseModel):
    """Pagination metadata."""
    total: int
    skip: int
    limit: int
    count: int

Step 2: In-memory data — app/data.py

from datetime import datetime, timedelta
from app.models import TaskInDB


def create_sample_tasks() -> list[TaskInDB]:
    """Generates sample data for the API."""
    base_date = datetime(2026, 3, 1, 9, 0, 0)
    return [
        TaskInDB(
            id=1, title="Set up the CI/CD pipeline",
            description="Implement GitHub Actions with testing and automatic deploys",
            status="in_progress", priority=5, due_date="2026-03-20",
            created_at=base_date, created_by="admin",
            internal_notes="Coordinate with the DevOps team", is_archived=False,
        ),
        TaskInDB(
            id=2, title="Write unit tests",
            description="Minimum 80% coverage for the core modules",
            status="pending", priority=4, due_date="2026-03-25",
            created_at=base_date + timedelta(hours=2), created_by="dev-lead",
            internal_notes=None, is_archived=False,
        ),
        TaskInDB(
            id=3, title="Document the API endpoints",
            description="Add docstrings and examples in OpenAPI",
            status="completed", priority=3,
            created_at=base_date + timedelta(days=1), updated_at=base_date + timedelta(days=3),
            created_by="dev-lead", internal_notes="Finished ahead of the deadline",
            is_archived=False,
        ),
        TaskInDB(
            id=4, title="Refactor the auth module",
            description="Split authentication and authorization logic",
            status="pending", priority=4, due_date="2026-04-01",
            created_at=base_date + timedelta(days=2), created_by="tech-lead",
            internal_notes="Needs a security review", is_archived=False,
        ),
        TaskInDB(
            id=5, title="Optimize the database queries",
            description="Cut response time on the listing endpoints",
            status="in_progress", priority=3,
            created_at=base_date + timedelta(days=3), created_by="dba",
            internal_notes=None, is_archived=False,
        ),
        TaskInDB(
            id=6, title="Migrate to Pydantic v2",
            description="Update every model from v1 to v2",
            status="completed", priority=2,
            created_at=base_date - timedelta(days=10), updated_at=base_date - timedelta(days=2),
            created_by="dev-lead", internal_notes="Migration finished with no issues",
            is_archived=True,
        ),
        TaskInDB(
            id=7, title="Implement rate limiting",
            description="Add per-IP rate limiting on the public endpoints",
            status="pending", priority=5, due_date="2026-03-18",
            created_at=base_date + timedelta(days=4), created_by="security-team",
            internal_notes="High priority after last week's incident",
            is_archived=False,
        ),
        TaskInDB(
            id=8, title="Add a health check endpoint",
            description="A /health endpoint for monitoring",
            status="completed", priority=1,
            created_at=base_date + timedelta(days=5), updated_at=base_date + timedelta(days=5, hours=2),
            created_by="system", internal_notes=None, is_archived=False,
        ),
    ]


tasks_db: list[TaskInDB] = create_sample_tasks()

Step 3: Response helpers — app/responses.py

from fastapi import Response
from fastapi.responses import JSONResponse
from app.models import ErrorResponse


API_VERSION = "3.0.0"


def add_standard_headers(response: Response, **extra_headers: str) -> None:
    """Adds the standard headers to any response."""
    response.headers["X-API-Version"] = API_VERSION
    for key, value in extra_headers.items():
        header_name = key.replace("_", "-")
        response.headers[header_name] = str(value)


def error_json(message: str, error_code: str, status_code: int = 400, detail: str | None = None) -> JSONResponse:
    """Builds an error JSONResponse with a consistent structure."""
    error = ErrorResponse(
        message=message,
        error_code=error_code,
        detail=detail,
    )
    response = JSONResponse(
        status_code=status_code,
        content=error.model_dump(),
    )
    response.headers["X-API-Version"] = API_VERSION
    response.headers["X-Error-Code"] = error_code
    return response

Step 4: The tasks router — app/routers/tasks.py

This is the main file. Every endpoint has professional responses:

from fastapi import APIRouter, HTTPException, Query, Path, Response
from fastapi.responses import StreamingResponse
from datetime import datetime
from typing import Optional, Literal

from app.models import (
    TaskCreate, TaskUpdate, TaskPublic, TaskAdmin,
    TaskSummary, TaskInDB, ErrorResponse, PaginationMeta,
)
from app.data import tasks_db
from app.responses import add_standard_headers, error_json

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


# --- Public listing ---

@router.get(
    "",
    response_model=list[TaskSummary],
    summary="List tasks (compact view)",
    responses={
        200: {
            "description": "A list of tasks in compact format",
            "content": {
                "application/json": {
                    "example": [
                        {"id": 1, "title": "Set up CI/CD", "status": "in_progress", "priority": 5},
                    ],
                },
            },
        },
    },
)
def list_tasks(
    response: Response,
    status: Optional[Literal["pending", "in_progress", "completed"]] = Query(
        default=None, description="Filter by status",
    ),
    priority: Optional[int] = Query(default=None, ge=1, le=5, description="Filter by priority"),
    search: Optional[str] = Query(default=None, min_length=1, max_length=100, description="Search in the title"),
    skip: int = Query(default=0, ge=0, description="Records to skip"),
    limit: int = Query(default=10, ge=1, le=100, description="Records per page"),
):
    """Lists tasks with filters and pagination. Returns the compact view (TaskSummary)."""
    results = [t for t in tasks_db if not t.is_archived]

    if status:
        results = [t for t in results if t.status == status]
    if priority is not None:
        results = [t for t in results if t.priority == priority]
    if search:
        term = search.lower()
        results = [t for t in results if term in t.title.lower()]

    total = len(results)
    paginated = results[skip : skip + limit]

    add_standard_headers(
        response,
        X_Total_Count=total,
        X_Skip=skip,
        X_Limit=limit,
    )

    return paginated


# --- Admin listing ---

@router.get(
    "/admin",
    response_model=list[TaskAdmin],
    summary="List tasks (admin view)",
    responses={
        200: {"description": "The complete list of tasks with every field"},
    },
)
def list_tasks_admin(
    response: Response,
    include_archived: bool = Query(default=False, description="Include archived tasks"),
):
    """Lists tasks with every field. Includes internal fields."""
    results = tasks_db if include_archived else [t for t in tasks_db if not t.is_archived]

    add_standard_headers(response, X_Total_Count=len(results))

    return results


# --- Public detail ---

@router.get(
    "/{task_id}",
    response_model=TaskPublic,
    summary="Get a task by ID",
    responses={
        200: {
            "description": "Task found",
            "content": {
                "application/json": {
                    "example": {
                        "id": 1, "title": "Set up CI/CD",
                        "description": "Implement GitHub Actions",
                        "status": "in_progress", "priority": 5,
                        "due_date": "2026-03-20",
                        "created_at": "2026-03-01T09:00:00",
                    },
                },
            },
        },
        404: {
            "description": "Task not found",
            "model": ErrorResponse,
        },
    },
)
def get_task(task_id: int = Path(ge=1, description="The task's ID")):
    """Returns a task by ID. The public view, with no internal fields."""
    task = next((t for t in tasks_db if t.id == task_id), None)
    if task is None:
        return error_json(
            message=f"Task with id {task_id} not found",
            error_code="TASK_NOT_FOUND",
            status_code=404,
        )
    return task


# --- Admin detail ---

@router.get(
    "/{task_id}/admin",
    response_model=TaskAdmin,
    summary="Get a task by ID (admin)",
    responses={
        200: {"description": "The task with every field"},
        404: {"model": ErrorResponse, "description": "Task not found"},
    },
)
def get_task_admin(task_id: int = Path(ge=1)):
    """Returns a task by ID with every field (the admin view)."""
    task = next((t for t in tasks_db if t.id == task_id), None)
    if task is None:
        return error_json(
            message=f"Task with id {task_id} not found",
            error_code="TASK_NOT_FOUND",
            status_code=404,
        )
    return task


# --- Create ---

@router.post(
    "",
    response_model=TaskPublic,
    status_code=201,
    summary="Create a task",
    responses={
        201: {"description": "Task created successfully"},
        422: {"description": "Invalid data"},
    },
)
def create_task(task: TaskCreate, response: Response):
    """Creates a new task. Returns the public view."""
    new_id = max((t.id for t in tasks_db), default=0) + 1

    task_in_db = TaskInDB(
        id=new_id,
        **task.model_dump(),
        created_at=datetime.now(),
        created_by="api-user",
        is_archived=False,
    )
    tasks_db.append(task_in_db)

    add_standard_headers(response)
    response.headers["Location"] = f"/tasks/{new_id}"

    return task_in_db


# --- Partial update ---

@router.patch(
    "/{task_id}",
    response_model=TaskPublic,
    response_model_exclude_unset=True,
    summary="Update a task partially",
    responses={
        200: {"description": "Task updated"},
        404: {"model": ErrorResponse, "description": "Task not found"},
        422: {"description": "Invalid update data"},
    },
)
def update_task(
    task_id: int = Path(ge=1),
    updates: TaskUpdate = ...,
):
    """Updates only the fields provided. Returns only the fields that changed."""
    task = next((t for t in tasks_db if t.id == task_id), None)
    if task is None:
        return error_json(
            message=f"Task with id {task_id} not found",
            error_code="TASK_NOT_FOUND",
            status_code=404,
        )

    update_data = updates.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(task, field, value)
    task.updated_at = datetime.now()

    return task


# --- Delete ---

@router.delete(
    "/{task_id}",
    status_code=200,
    summary="Delete a task",
    responses={
        200: {
            "description": "Task deleted",
            "content": {
                "application/json": {
                    "example": {"message": "Task deleted", "deleted_id": 1},
                },
            },
        },
        404: {"model": ErrorResponse, "description": "Task not found"},
    },
)
def delete_task(task_id: int = Path(ge=1)):
    """Deletes a task by ID."""
    task = next((t for t in tasks_db if t.id == task_id), None)
    if task is None:
        return error_json(
            message=f"Task with id {task_id} not found",
            error_code="TASK_NOT_FOUND",
            status_code=404,
        )

    tasks_db.remove(task)
    return {"message": "Task deleted", "deleted_id": task_id}


# --- CSV export ---

def tasks_csv_generator(tasks: list[TaskInDB]):
    """Generator for streaming tasks as CSV."""
    yield "id,title,status,priority,due_date,created_at\n"
    for task in tasks:
        due = task.due_date or ""
        created = task.created_at.isoformat()
        title = task.title.replace(",", ";")
        yield f"{task.id},{title},{task.status},{task.priority},{due},{created}\n"


@router.get(
    "/export/csv",
    summary="Export tasks as CSV",
    responses={
        200: {
            "description": "A CSV file with the tasks",
            "content": {
                "text/csv": {
                    "example": "id,title,status,priority,due_date,created_at\n1,Setup CI/CD,in_progress,5,2026-03-20,2026-03-01T09:00:00\n",
                },
            },
        },
    },
)
def export_tasks_csv(
    status: Optional[Literal["pending", "in_progress", "completed"]] = Query(
        default=None, description="Filter by status before exporting",
    ),
):
    """Exports tasks as streaming CSV. Optionally filters by status."""
    data = [t for t in tasks_db if not t.is_archived]
    if status:
        data = [t for t in data if t.status == status]

    filename = f"tasks_{status or 'all'}_{datetime.now().strftime('%Y%m%d')}.csv"

    return StreamingResponse(
        content=tasks_csv_generator(data),
        media_type="text/csv",
        headers={
            "Content-Disposition": f"attachment; filename={filename}",
            "X-Export-Count": str(len(data)),
            "X-API-Version": "3.0.0",
        },
    )


# --- Statistics ---

@router.get(
    "/stats/summary",
    summary="Task statistics",
    responses={
        200: {
            "description": "Aggregated statistics",
            "content": {
                "application/json": {
                    "example": {
                        "total": 8,
                        "by_status": {"pending": 3, "in_progress": 2, "completed": 3},
                        "by_priority": {"1": 1, "5": 2},
                        "archived": 1,
                    },
                },
            },
        },
    },
)
def task_stats(response: Response):
    """Aggregated statistics across all tasks."""
    total = len(tasks_db)
    archived = sum(1 for t in tasks_db if t.is_archived)
    active = total - archived

    by_status: dict[str, int] = {}
    by_priority: dict[str, int] = {}

    for task in tasks_db:
        by_status[task.status] = by_status.get(task.status, 0) + 1
        key = str(task.priority)
        by_priority[key] = by_priority.get(key, 0) + 1

    add_standard_headers(response, X_Total_Count=total)

    return {
        "total": total,
        "active": active,
        "archived": archived,
        "by_status": by_status,
        "by_priority": by_priority,
    }

Step 5: The main app — app/main.py

from fastapi import FastAPI
from app.routers import tasks

app = FastAPI(
    title="Task Manager API",
    description="A task management API with professional responses. Module 3 — FastAPI Advanced Features.",
    version="3.0.0",
)

app.include_router(tasks.router)


@app.get("/", tags=["General"])
def root():
    return {
        "service": "Task Manager API",
        "version": "3.0.0",
        "modules": "DI + Routers + Response Models",
        "docs": "/docs",
    }

Step 6: Run it and verify

uvicorn app.main:app --reload

Check 1: The compact listing (TaskSummary)

curl -s http://127.0.0.1:8000/tasks | python -m json.tool
[
    {"id": 1, "title": "Set up the CI/CD pipeline", "status": "in_progress", "priority": 5},
    {"id": 2, "title": "Write unit tests", "status": "pending", "priority": 4}
]

Only 4 fields per task. No description, no created_at, no internal fields.

Check 2: The admin listing (TaskAdmin)

curl -s "http://127.0.0.1:8000/tasks/admin?include_archived=true" | python -m json.tool

Returns every task (including the archived ones) with every field: created_by, internal_notes, is_archived.

Check 3: Custom headers

curl -v http://127.0.0.1:8000/tasks 2>&1 | grep -i "x-"
< x-api-version: 3.0.0
< x-total-count: 7
< x-skip: 0
< x-limit: 10

Check 4: Filters with pagination

curl -s "http://127.0.0.1:8000/tasks?status=pending&skip=0&limit=2" | python -m json.tool

Returns only pending tasks, paginated.

Check 5: Public detail vs admin detail

curl -s http://127.0.0.1:8000/tasks/1 | python -m json.tool
# → Public view: no internal_notes, created_by, is_archived

curl -s http://127.0.0.1:8000/tasks/1/admin | python -m json.tool
# → Admin view: every field including internal_notes

Check 6: A consistent error response

curl -s http://127.0.0.1:8000/tasks/999 | python -m json.tool
{
    "status": "error",
    "message": "Task with id 999 not found",
    "detail": null,
    "error_code": "TASK_NOT_FOUND"
}

With an X-Error-Code: TASK_NOT_FOUND header and HTTP status 404.

Check 7: PATCH with exclude_unset

curl -s -X PATCH http://127.0.0.1:8000/tasks/1 \
  -H "Content-Type: application/json" \
  -d '{"status": "completed"}' | python -m json.tool

The response includes only the fields that hold a value — it doesn't return description: null or due_date: null if they weren't part of the model's original setup.

Check 8: CSV export

curl http://127.0.0.1:8000/tasks/export/csv
id,title,status,priority,due_date,created_at
1,Set up the CI/CD pipeline,in_progress,5,2026-03-20,2026-03-01T09:00:00
2,Write unit tests,pending,4,2026-03-25,2026-03-01T11:00:00
...

Check 9: Filtered CSV export

curl -v "http://127.0.0.1:8000/tasks/export/csv?status=completed" 2>&1 | grep -i "x-export"
# x-export-count: 2

It exports only completed tasks. The X-Export-Count header confirms how many went out.

Check 10: Statistics

curl -s http://127.0.0.1:8000/tasks/stats/summary | python -m json.tool
{
    "total": 8,
    "active": 7,
    "archived": 1,
    "by_status": {"in_progress": 2, "pending": 3, "completed": 3},
    "by_priority": {"5": 2, "4": 2, "3": 2, "2": 1, "1": 1}
}

Check 11: Creating a task with a Location header

curl -v -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "New task", "priority": 3}' 2>&1 | grep -i "location"
# location: /tasks/9

The Location header points to the URL of the resource you just created.

Check 12: The documentation in /docs

Open http://127.0.0.1:8000/docs. Expand each endpoint and verify:

  • Every endpoint shows its possible responses (200, 404, 422)
  • The response schemas are documented with examples
  • The query parameters have descriptions
  • The CSV export documents the text/csv content-type

Completeness checklist

Schemas:
- [ ] TaskSummary with 4 fields (id, title, status, priority)
- [ ] TaskPublic with the base fields + id + created_at (no internal ones)
- [ ] TaskAdmin with every field including the internal ones
- [ ] TaskCreate for POST
- [ ] TaskUpdate with everything Optional for PATCH
- [ ] ErrorResponse with status, message, error_code
- [ ] TaskInDB as the complete storage model

Endpoints — Listing:
- [ ] GET /tasks returns list[TaskSummary]
- [ ] GET /tasks/admin returns list[TaskAdmin]
- [ ] Filters: status, priority, search
- [ ] Pagination: skip, limit
- [ ] Headers: X-Total-Count, X-Skip, X-Limit

Endpoints — Detail:
- [ ] GET /tasks/{id} returns TaskPublic
- [ ] GET /tasks/{id}/admin returns TaskAdmin
- [ ] 404 with ErrorResponse for a task that doesn't exist

Endpoints — Mutation:
- [ ] POST /tasks returns TaskPublic + a Location header
- [ ] PATCH /tasks/{id} uses response_model_exclude_unset
- [ ] DELETE /tasks/{id} returns a confirmation message

Endpoints — Export and Stats:
- [ ] GET /tasks/export/csv returns a StreamingResponse with CSV
- [ ] The CSV export accepts a status filter
- [ ] An X-Export-Count header on the export
- [ ] GET /tasks/stats/summary returns aggregated statistics

Responses:
- [ ] Every error response uses ErrorResponse
- [ ] An X-API-Version header on every response
- [ ] An X-Error-Code header on error responses
- [ ] The responses parameter documenting every possible response

OpenAPI documentation:
- [ ] /docs shows the TaskSummary, TaskPublic, TaskAdmin schemas
- [ ] Every endpoint documents its possible status codes
- [ ] Examples on the main responses
- [ ] The CSV export documents the text/csv content-type

Troubleshooting

Problem 1: GET /tasks/export/csv returns a 422 instead of the CSV

Cause: FastAPI reads /tasks/export as /tasks/{task_id} with task_id="export". The endpoint with the path parameter is capturing the route first.

Fix: Declare the fixed routes (/export/csv, /stats/summary, /admin) BEFORE the route with {task_id}:

# ✅ The right order in the router
@router.get("/export/csv")    # fixed — first
@router.get("/stats/summary") # fixed — first
@router.get("/admin")         # fixed — first
@router.get("/{task_id}")     # parameterized — last

Problem 2: TaskPublic doesn't filter TaskInDB's fields

Cause: response_model=TaskPublic only filters if FastAPI can build a TaskPublic from the data. If you return a dict with extra fields, response_model excludes them. But if there are fields that don't match, it can fail.

Fix: Make sure TaskPublic has ConfigDict(from_attributes=True) and that you return the TaskInDB object directly (not a partial dict):

class TaskPublic(TaskBase):
    model_config = ConfigDict(from_attributes=True)
    id: int
    created_at: datetime

@router.get("/tasks/{id}", response_model=TaskPublic)
def get_task(id: int):
    return task_in_db  # FastAPI filters it down to TaskPublic automatically

Problem 3: The CSV repeats the header row between chunked responses

Cause: The generator is emitting the header in every chunk, not just at the start.

Fix: Check that the header's yield sits outside the loop:

def csv_gen(data):
    yield "id,title,status\n"  # Just once, outside the loop
    for item in data:
        yield f"{item.id},{item.title},{item.status}\n"

Problem 4: PATCH returns every field despite exclude_unset

Cause: response_model_exclude_unset works on the fields of the model that was used to create the instance. If you rebuild the object with every field, exclude_unset has no effect because everything counts as "set".

Fix: Don't rebuild the object — modify the existing one with setattr:

# ❌ Rebuilding loses the "unset" information
task = TaskInDB(**{**old_task.model_dump(), **updates})

# ✅ Modifying in place preserves it
for field, value in updates.items():
    setattr(task, field, value)

Problem 5: The custom X- headers don't show up in the browser

Cause: CORS doesn't expose custom headers by default. The browser receives them but JavaScript can't read them.

Fix: Configure CORS to expose the headers (covered in FastAPI Fundamentals, but the quick fix is):

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    expose_headers=["X-Total-Count", "X-API-Version", "X-Export-Count", "X-Error-Code"],
)

Problem 6: error_json returns a 200 instead of the right status code

Cause: If you return a JSONResponse from an endpoint that has a response_model, FastAPI honors the JSONResponse's status code. But if you accidentally return a dict shaped like an error, FastAPI stamps it with status 200.

Fix: Always use an explicit JSONResponse or HTTPException for errors — never return an error dict as a normal return value:

# ❌ Returns an error dict with status 200
return {"status": "error", "message": "Not found"}

# ✅ Returns a JSONResponse with status 404
return JSONResponse(status_code=404, content={"status": "error", "message": "Not found"})

# ✅ Or use HTTPException
raise HTTPException(status_code=404, detail="Not found")

Problem 7: The stats endpoint counts archived tasks

Cause: The stats endpoint doesn't filter out archived tasks, so it shows the wrong totals for the active context.

Fix: Document clearly that stats include archived tasks (if that's intentional), or filter according to the use case:

# Stats include EVERYTHING (archived and active) — document it
return {
    "total": len(tasks_db),
    "active": sum(1 for t in tasks_db if not t.is_archived),
    "archived": sum(1 for t in tasks_db if t.is_archived),
}

Patterns you applied

1. Schemas as a public contract

Every consumer gets its own schema. TaskSummary for quick listings, TaskPublic for normal views, TaskAdmin for the back office. The same TaskInDB object serializes differently depending on the context — response_model does the magic.

2. Consistent error responses

Every error uses ErrorResponse with the same structure: status, message, error_code. The client can build error handling around error_code without parsing strings.

3. Headers as metadata

The data goes in the body, the metadata about the response goes in headers. X-Total-Count for pagination, X-Export-Count for exports, X-API-Version for versioning. The body stays clean.

4. Streaming for exports

The CSV export uses StreamingResponse with a generator — constant memory no matter how many tasks there are. The X-Export-Count header tells the client how many records are in the file before it finishes downloading.

5. OpenAPI as living documentation

The responses parameter in every decorator documents all the possible responses. A developer who opens /docs knows exactly what they can get from each endpoint and each status code, with real examples.

Summary

In this project you brought together every tool from Module 3:

  • Differentiated schemas: TaskSummary, TaskPublic, TaskAdmin — the same data, three views depending on the consumer
  • response_model_exclude_unset on PATCH — the response only includes the relevant fields
  • StreamingResponse with a generator for CSV export — constant memory
  • Custom headers: X-Total-Count, X-API-Version, X-Export-Count, X-Error-Code, Location
  • Consistent error responses with the ErrorResponse schema and the error_json helper
  • OpenAPI documentation with the responses parameter — every endpoint documents its possible responses
  • Response helpers (add_standard_headers, error_json) encapsulate the repetitive logic
  • Filters on export — the CSV accepts filter parameters before exporting

Your API doesn't just "return data" anymore — it returns exactly what each consumer needs, in the right format, with useful metadata in the headers, and complete documentation in /docs.

Additional resources

  1. FastAPI - Response Model — response_model, exclude, include, exclude_unset
  2. FastAPI - Additional Responses — Documenting multiple responses with the responses parameter
  3. FastAPI - Custom Response — JSONResponse, StreamingResponse, FileResponse
  4. FastAPI - Response Headers — Custom headers on responses
  5. Pydantic v2 - Serialization — model_dump, include, exclude, exclude_unset
  6. OpenAPI Specification - Responses — How responses are documented in OpenAPI

What's next?

Your API returns data professionally. But there are operations that shouldn't block the response to the client. When a task gets created, maybe you want to send a notification email. When it gets completed, maybe you want to update a dashboard. Those operations take time — and the client shouldn't have to wait.

In Module 4 (Background Tasks) you'll learn to run operations after the response goes out:

Module 3 (now):                          Module 4 (next):
──────────────────                        ─────────────────────
POST /tasks → creates task → responds  →  POST /tasks → responds → sends email in the background
GET /tasks → returns everything at once → GET /tasks → returns → clears cache in the background
Synchronous operations                 →  BackgroundTasks for fire-and-forget
No post-response processing            →  Celery/RQ for heavy tasks

Module 3 complete. You've mastered total control over responses. Next stop: Background Tasks, where your API does things after it replies to the client.

← PreviousCustom Responses and HeadersNext →Introduction to Module 4: Background Tasks