Module 6: Project — Advanced Task Manager API

CRUD Endpoints, Responses, and Background Tasks

Next →

Overview

With the architecture in place (models, data, dependencies), now you implement the endpoints: a full CRUD with filters, statistics, CSV export, and background tasks for email and auditing.

Tasks Router: the full CRUD

Create app/routers/tasks.py:

import csv
import io
from collections import Counter
from fastapi import APIRouter, HTTPException, Depends, BackgroundTasks
from fastapi.responses import StreamingResponse
from app.models import (
    TaskCreate, TaskUpdate, TaskPatch,
    TaskResponse, TaskSummary, TaskStats,
    Status, Priority
)
from app.data import tasks_db, generate_id, get_current_timestamp
from app.dependencies import (
    pagination_params, get_task_or_404,
    filter_params, apply_filters
)
from app.websocket_manager import ws_manager
from app.background.notifications import send_email_notification, log_audit_event

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


@router.post("/", status_code=201, response_model=TaskResponse)
async def create_task(
    task_data: TaskCreate,
    background_tasks: BackgroundTasks
):
    new_task = {
        "id": generate_id(),
        "title": task_data.title,
        "description": task_data.description,
        "status": task_data.status.value,
        "priority": task_data.priority.value,
        "assignee": task_data.assignee,
        "tags": task_data.tags,
        "attachments": [],
        "created_at": get_current_timestamp(),
        "updated_at": get_current_timestamp()
    }
    tasks_db.append(new_task)

    background_tasks.add_task(
        send_email_notification,
        event="task_created",
        task_title=new_task["title"],
        task_id=new_task["id"]
    )
    background_tasks.add_task(
        log_audit_event,
        action="create_task",
        details={"task_id": new_task["id"], "title": new_task["title"]}
    )

    await ws_manager.broadcast_event("task_created", {
        "task_id": new_task["id"],
        "title": new_task["title"],
        "priority": new_task["priority"],
        "assignee": new_task["assignee"]
    })

    return new_task


@router.get("/", response_model=list[TaskSummary])
async def list_tasks(
    pagination: dict = Depends(pagination_params),
    filters: dict = Depends(filter_params)
):
    filtered = apply_filters(tasks_db, filters)
    skip = pagination["skip"]
    limit = pagination["limit"]
    page = filtered[skip:skip + limit]

    return [
        {
            **task,
            "attachments_count": len(task.get("attachments", []))
        }
        for task in page
    ]


@router.get("/stats", response_model=TaskStats)
async def task_stats():
    total = len(tasks_db)
    if total == 0:
        return TaskStats(
            total=0,
            by_status={},
            by_priority={},
            completion_rate=0.0,
            with_attachments=0
        )

    status_counts = Counter(t["status"] for t in tasks_db)
    priority_counts = Counter(t["priority"] for t in tasks_db)
    completed = status_counts.get("completed", 0)
    with_attachments = sum(1 for t in tasks_db if t.get("attachments"))

    return TaskStats(
        total=total,
        by_status=dict(status_counts),
        by_priority=dict(priority_counts),
        completion_rate=round((completed / total) * 100, 1),
        with_attachments=with_attachments
    )


@router.get("/export/csv")
async def export_csv():
    output = io.StringIO()
    writer = csv.writer(output)

    writer.writerow([
        "id", "title", "description", "status",
        "priority", "assignee", "tags", "attachments_count",
        "created_at", "updated_at"
    ])

    for task in tasks_db:
        writer.writerow([
            task["id"],
            task["title"],
            task["description"],
            task["status"],
            task["priority"],
            task.get("assignee", ""),
            ", ".join(task.get("tags", [])),
            len(task.get("attachments", [])),
            task["created_at"],
            task["updated_at"]
        ])

    output.seek(0)

    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=tasks_export.csv"}
    )


@router.get("/{task_id}", response_model=TaskResponse)
async def get_task(task: dict = Depends(get_task_or_404)):
    return task


@router.put("/{task_id}", response_model=TaskResponse)
async def update_task(
    task_data: TaskUpdate,
    task: dict = Depends(get_task_or_404),
    background_tasks: BackgroundTasks = BackgroundTasks()
):
    task["title"] = task_data.title
    task["description"] = task_data.description
    task["status"] = task_data.status.value
    task["priority"] = task_data.priority.value
    task["assignee"] = task_data.assignee
    task["tags"] = task_data.tags
    task["updated_at"] = get_current_timestamp()

    background_tasks.add_task(
        log_audit_event,
        action="update_task",
        details={"task_id": task["id"], "title": task["title"]}
    )

    await ws_manager.broadcast_event("task_updated", {
        "task_id": task["id"],
        "title": task["title"],
        "status": task["status"]
    })

    return task


@router.patch("/{task_id}", response_model=TaskResponse)
async def patch_task(
    task_data: TaskPatch,
    task: dict = Depends(get_task_or_404),
    background_tasks: BackgroundTasks = BackgroundTasks()
):
    update_data = task_data.model_dump(exclude_unset=True)
    old_status = task.get("status")

    for field, value in update_data.items():
        if hasattr(value, 'value'):
            task[field] = value.value
        else:
            task[field] = value
    task["updated_at"] = get_current_timestamp()

    new_status = task.get("status")

    if old_status != new_status and new_status == "completed":
        await ws_manager.broadcast_event("task_completed", {
            "task_id": task["id"],
            "title": task["title"],
            "completed_at": task["updated_at"]
        })

        background_tasks.add_task(
            send_email_notification,
            event="task_completed",
            task_title=task["title"],
            task_id=task["id"]
        )
    else:
        await ws_manager.broadcast_event("task_patched", {
            "task_id": task["id"],
            "fields_updated": list(update_data.keys())
        })

    background_tasks.add_task(
        log_audit_event,
        action="patch_task",
        details={
            "task_id": task["id"],
            "fields": list(update_data.keys())
        }
    )

    return task


@router.delete("/{task_id}", status_code=204)
async def delete_task(
    task: dict = Depends(get_task_or_404),
    background_tasks: BackgroundTasks = BackgroundTasks()
):
    task_info = {"id": task["id"], "title": task["title"]}
    tasks_db.remove(task)

    background_tasks.add_task(
        log_audit_event,
        action="delete_task",
        details=task_info
    )

    await ws_manager.broadcast_event("task_deleted", task_info)

Key points of the implementation

Filters + pagination combined

The GET /tasks/ endpoint uses two dependencies:

@router.get("/")
async def list_tasks(
    pagination: dict = Depends(pagination_params),
    filters: dict = Depends(filter_params)
):

It filters first, then paginates. The order matters — you don't want to paginate before filtering.

CSV export with StreamingResponse

@router.get("/export/csv")
async def export_csv():
    output = io.StringIO()
    # ... write CSV ...
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=tasks_export.csv"}
    )

Content-Disposition: attachment makes the browser download the file instead of displaying it.

Background tasks in the CRUD

Every CRUD operation fires:

  1. An email notification (simulated with logging) — for the important operations
  2. An audit log entry — for every operation
  3. A WebSocket broadcast — for the real-time notification

Background tasks run after the response is sent, so the client doesn't wait.

Task completion detection

In PATCH, we detect when a task changes to "completed":

old_status = task.get("status")
# ... apply updates ...
new_status = task.get("status")

if old_status != new_status and new_status == "completed":
    # special event

This lets you send a task_completed event that's distinct from task_patched.

A special route: /tasks/stats before /tasks/{task_id}

Notice the order in the router: @router.get("/stats") comes before @router.get("/{task_id}"). If it were the other way around, FastAPI would try to read "stats" as a task_id.

Partial verification

# Start the server
uvicorn app.main:app --reload

# Create a task
curl -s -X POST http://localhost:8000/tasks/ \
  -H "Content-Type: application/json" \
  -d '{"title": "Test task", "priority": "high", "tags": ["test"]}' | python -m json.tool

# List tasks (includes the 5 seeded ones + the new one)
curl -s "http://localhost:8000/tasks/" | python -m json.tool

# Filter by status
curl -s "http://localhost:8000/tasks/?status=pending" | python -m json.tool

# Filter by priority
curl -s "http://localhost:8000/tasks/?priority=critical" | python -m json.tool

# Search by text
curl -s "http://localhost:8000/tasks/?search=websocket" | python -m json.tool

# Statistics
curl -s "http://localhost:8000/tasks/stats" | python -m json.tool

# Export CSV
curl -s "http://localhost:8000/tasks/export/csv" -o tasks.csv
cat tasks.csv

# Complete a task
curl -s -X PATCH http://localhost:8000/tasks/3 \
  -H "Content-Type: application/json" \
  -d '{"status": "completed"}' | python -m json.tool

# Delete a task
curl -s -X DELETE http://localhost:8000/tasks/4 -w "\nHTTP: %{http_code}\n"

Check the server logs — you should see the email and audit notifications.

Troubleshooting

/tasks/stats returns 422 or gets read as a task_id

Make sure @router.get("/stats") is defined before @router.get("/{task_id}") in the code. FastAPI evaluates routes in order.

The background tasks don't run

Check that BackgroundTasks is injected as an endpoint parameter, not created manually. FastAPI needs to inject it for it to work.

A filter returns an empty list when there is data

Check that the filter value matches the Enum values exactly. status=pending works, status=Pending doesn't.

The CSV export shows HTML instead of downloading

Make sure you include Content-Disposition: attachment in the StreamingResponse headers.

The audit log is empty after operations

Check that from app.data import audit_log in notifications.py imports the same list. Python shares the reference if you use the right import.

Resources

  1. FastAPI BackgroundTasks — The official docs
  2. StreamingResponse — Streaming responses
  3. FastAPI Dependencies — Injection patterns
  4. Pydantic model_dump — Serialization

What's next?

In Capsule 04 you add the WebSocket endpoint, the file uploads router, and the logging middleware. It's the final layer of advanced features.

← PreviousArchitecture: Models, Data, and DependenciesNext →WebSocket, File Uploads, and Middleware