Module 4: Background Tasks

Project: Background Processing for the Task Manager API

Next →

Project overview

In this project you integrate everything you learned in Module 4: BackgroundTasks for post-response operations, error handling with try/except + logging, in-memory status tracking, and the notification and cleanup patterns. Your Task Manager API goes from "respond and we're done" to "respond immediately and process afterward" — the way production APIs do it.

The result is an API that, when a task is created, sends a "notification" (a structured log) to the team; when a task is completed, records it in an audit log; lets you clean up old completed tasks with progress tracking; and has an endpoint for checking the state of background operations. All without the client waiting a single extra second.

Project objectives

By the end of this project you'll have:

  • ✅ A simulated notification (a log) when a task is created
  • ✅ A notification when a task is completed (status changes to "completed")
  • ✅ A background audit log for write operations (create, update, delete)
  • ✅ A cleanup endpoint that deletes tasks completed more than N days ago
  • ✅ Status tracking for background operations
  • ✅ A GET /background/status endpoint for checking the state of operations
  • ✅ Robust error handling — background failures don't affect the client
  • ✅ Structured logging for monitoring

Project structure

fastapi-advanced/
├── app/
│   ├── __init__.py
│   ├── main.py              ← The main app
│   ├── models.py            ← Pydantic schemas
│   ├── data.py              ← In-memory data
│   ├── background.py        ← Background task functions + the tracker
│   └── routers/
│       ├── __init__.py
│       └── tasks.py         ← The tasks router with background processing
└── requirements.txt

Step 1: Pydantic schemas — app/models.py

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


class TaskBase(BaseModel):
    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)


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


class TaskUpdate(BaseModel):
    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)


class TaskPublic(TaskBase):
    id: int
    assignee_email: Optional[str] = None
    created_at: datetime
    updated_at: Optional[datetime] = None


class TaskInDB(TaskPublic):
    created_by: str = "api-user"
    internal_notes: Optional[str] = None


class BackgroundJobStatus(BaseModel):
    job_id: str
    job_type: str
    status: Literal["pending", "running", "completed", "failed"]
    result: Optional[str] = None
    error: Optional[str] = None
    created_at: datetime
    completed_at: Optional[datetime] = None

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

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


def create_sample_tasks() -> list[TaskInDB]:
    base = datetime(2026, 3, 1, 9, 0, 0)
    return [
        TaskInDB(
            id=1, title="Set up CI/CD pipeline",
            description="GitHub Actions with automated testing",
            status="in_progress", priority=5,
            assignee_email="devops@team.com",
            created_at=base, created_by="admin",
        ),
        TaskInDB(
            id=2, title="Write unit tests",
            status="pending", priority=4,
            assignee_email="dev@team.com",
            created_at=base + timedelta(hours=2), created_by="dev-lead",
        ),
        TaskInDB(
            id=3, title="Document API endpoints",
            status="completed", priority=3,
            created_at=base - timedelta(days=35),
            updated_at=base - timedelta(days=30),
            created_by="dev-lead",
        ),
        TaskInDB(
            id=4, title="Refactor the auth module",
            status="pending", priority=4,
            assignee_email="security@team.com",
            created_at=base + timedelta(days=2), created_by="tech-lead",
        ),
        TaskInDB(
            id=5, title="Optimize SQL queries",
            status="completed", priority=3,
            created_at=base - timedelta(days=40),
            updated_at=base - timedelta(days=32),
            created_by="dba",
        ),
        TaskInDB(
            id=6, title="Migrate to Pydantic v2",
            status="completed", priority=2,
            created_at=base - timedelta(days=60),
            updated_at=base - timedelta(days=45),
            created_by="dev-lead",
        ),
        TaskInDB(
            id=7, title="Implement rate limiting",
            status="in_progress", priority=5,
            assignee_email="security@team.com",
            created_at=base + timedelta(days=4), created_by="security-team",
        ),
        TaskInDB(
            id=8, title="Add a health check",
            status="completed", priority=1,
            created_at=base - timedelta(days=20),
            updated_at=base - timedelta(days=18),
            created_by="system",
        ),
    ]


tasks_db: list[TaskInDB] = create_sample_tasks()

Step 3: Background task functions — app/background.py

This module holds every background function and the status tracker:

import logging
import time
import uuid
import functools
from datetime import datetime, timedelta
from typing import Optional

from app.models import BackgroundJobStatus

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger("background")

job_tracker: dict[str, BackgroundJobStatus] = {}

AUDIT_LOG: list[dict] = []


def safe_task(func):
    """A decorator that wraps background tasks with error handling and logging."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        name = func.__name__
        start = time.time()
        logger.info(f"[START] {name}")
        try:
            result = func(*args, **kwargs)
            elapsed = round(time.time() - start, 2)
            logger.info(f"[DONE] {name} ({elapsed}s)")
            return result
        except Exception as e:
            elapsed = round(time.time() - start, 2)
            logger.error(f"[ERROR] {name} ({elapsed}s): {type(e).__name__}: {e}")
    return wrapper


def create_job(job_type: str) -> str:
    """Registers a new job in the tracker and returns its job_id."""
    job_id = str(uuid.uuid4())[:8]
    job_tracker[job_id] = BackgroundJobStatus(
        job_id=job_id,
        job_type=job_type,
        status="pending",
        created_at=datetime.now(),
    )
    return job_id


def complete_job(job_id: str, result: str):
    """Marks a job as completed."""
    if job_id in job_tracker:
        job_tracker[job_id].status = "completed"
        job_tracker[job_id].result = result
        job_tracker[job_id].completed_at = datetime.now()


def fail_job(job_id: str, error: str):
    """Marks a job as failed."""
    if job_id in job_tracker:
        job_tracker[job_id].status = "failed"
        job_tracker[job_id].error = error
        job_tracker[job_id].completed_at = datetime.now()


@safe_task
def send_notification(email: str, subject: str, body: str):
    """Simulates sending a notification email."""
    time.sleep(1)
    logger.info(f"[NOTIFICATION] To: {email} | Subject: {subject} | Body: {body}")


@safe_task
def write_audit_log(
    action: str,
    resource_type: str,
    resource_id: int,
    details: Optional[str] = None,
):
    """Writes an entry to the audit log."""
    entry = {
        "timestamp": datetime.now().isoformat(),
        "action": action,
        "resource_type": resource_type,
        "resource_id": resource_id,
        "details": details,
    }
    AUDIT_LOG.append(entry)
    logger.info(f"[AUDIT] {action} {resource_type}:{resource_id} — {details or 'no details'}")


def cleanup_completed_tasks(
    tasks_db: list,
    days_old: int,
    job_id: str,
):
    """Deletes tasks completed more than N days ago. With status tracking."""
    try:
        job_tracker[job_id].status = "running"
        logger.info(f"[CLEANUP] Starting — removing tasks completed > {days_old} days ago")

        time.sleep(1)

        cutoff = datetime.now() - timedelta(days=days_old)
        to_remove = [
            t for t in tasks_db
            if t.status == "completed" and t.updated_at and t.updated_at < cutoff
        ]

        removed_count = 0
        for task in to_remove:
            tasks_db.remove(task)
            removed_count += 1
            logger.info(f"[CLEANUP] Removed task #{task.id}: '{task.title}' (completed {task.updated_at})")

        result = f"Removed {removed_count} completed tasks older than {days_old} days"
        complete_job(job_id, result)
        logger.info(f"[CLEANUP] {result}")

    except Exception as e:
        fail_job(job_id, str(e))
        logger.error(f"[CLEANUP ERROR] {type(e).__name__}: {e}")


def cleanup_old_jobs(max_age_minutes: int = 60):
    """Clears old completed/failed jobs out of the job tracker."""
    cutoff = datetime.now() - timedelta(minutes=max_age_minutes)
    expired = [
        jid for jid, job in job_tracker.items()
        if job.completed_at and job.completed_at < cutoff
    ]
    for jid in expired:
        del job_tracker[jid]
    if expired:
        logger.info(f"[TRACKER CLEANUP] Removed {len(expired)} old job entries")

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

from fastapi import APIRouter, HTTPException, BackgroundTasks, Query, Path, Response
from datetime import datetime
from typing import Optional, Literal

from app.models import TaskCreate, TaskUpdate, TaskPublic, TaskInDB, BackgroundJobStatus
from app.data import tasks_db
from app.background import (
    send_notification,
    write_audit_log,
    cleanup_completed_tasks,
    cleanup_old_jobs,
    create_job,
    job_tracker,
    AUDIT_LOG,
)

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


# --- Listing ---

@router.get("", response_model=list[TaskPublic])
def list_tasks(
    response: Response,
    status: Optional[Literal["pending", "in_progress", "completed"]] = Query(default=None),
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=10, ge=1, le=100),
):
    """Lists tasks with filters and pagination."""
    results = list(tasks_db)
    if status:
        results = [t for t in results if t.status == status]

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

    response.headers["X-Total-Count"] = str(total)
    return paginated


# --- Detail ---

@router.get("/{task_id}", response_model=TaskPublic)
def get_task(task_id: int = Path(ge=1)):
    """Gets a task by ID."""
    task = next((t for t in tasks_db if t.id == task_id), None)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return task


# --- Create (with background tasks) ---

@router.post("", response_model=TaskPublic, status_code=201)
def create_task(task: TaskCreate, background_tasks: BackgroundTasks, response: Response):
    """
    Creates a new task.

    Background tasks:
    - Sends a notification if there's an assignee_email
    - Records it in the audit log
    """
    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",
    )
    tasks_db.append(task_in_db)

    if task.assignee_email:
        background_tasks.add_task(
            send_notification,
            email=task.assignee_email,
            subject=f"New task assigned: {task.title}",
            body=f"You were assigned task #{new_id} with priority {task.priority}.",
        )

    background_tasks.add_task(
        write_audit_log,
        action="CREATE",
        resource_type="task",
        resource_id=new_id,
        details=f"Title: '{task.title}', Priority: {task.priority}",
    )

    response.headers["Location"] = f"/tasks/{new_id}"
    return task_in_db


# --- Update (with conditional background tasks) ---

@router.patch("/{task_id}", response_model=TaskPublic)
def update_task(
    task_id: int,
    updates: TaskUpdate,
    background_tasks: BackgroundTasks,
):
    """
    Partially updates a task.

    Background tasks:
    - If the status changes to 'completed', sends a notification
    - Records it in the audit log
    """
    task = next((t for t in tasks_db if t.id == task_id), None)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")

    old_status = task.status
    update_data = updates.model_dump(exclude_unset=True)

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

    if task.status == "completed" and old_status != "completed":
        if task.assignee_email:
            background_tasks.add_task(
                send_notification,
                email=task.assignee_email,
                subject=f"Task completed: {task.title}",
                body=f"Task #{task_id} was marked as completed.",
            )

        background_tasks.add_task(
            send_notification,
            email="team@example.com",
            subject=f"Task #{task_id} completed",
            body=f"'{task.title}' was completed.",
        )

    background_tasks.add_task(
        write_audit_log,
        action="UPDATE",
        resource_type="task",
        resource_id=task_id,
        details=f"Fields updated: {list(update_data.keys())}",
    )

    return task


# --- Delete (with background tasks) ---

@router.delete("/{task_id}", status_code=200)
def delete_task(task_id: int, background_tasks: BackgroundTasks):
    """
    Deletes a task.

    Background tasks:
    - Records it in the audit log
    """
    task = next((t for t in tasks_db if t.id == task_id), None)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")

    tasks_db.remove(task)

    background_tasks.add_task(
        write_audit_log,
        action="DELETE",
        resource_type="task",
        resource_id=task_id,
        details=f"Deleted task: '{task.title}'",
    )

    return {"message": f"Task {task_id} deleted", "deleted_id": task_id}


# --- Cleanup endpoint ---

@router.post("/cleanup", status_code=202)
def trigger_cleanup(
    background_tasks: BackgroundTasks,
    days_old: int = Query(default=30, ge=1, le=365, description="Delete tasks completed more than N days ago"),
):
    """
    Starts the cleanup of old completed tasks in the background.

    Returns a job_id you can use to check the status with GET /background/status/{job_id}.
    """
    job_id = create_job("cleanup_completed_tasks")

    background_tasks.add_task(cleanup_completed_tasks, tasks_db, days_old, job_id)

    background_tasks.add_task(cleanup_old_jobs, 60)

    return {
        "message": "Cleanup started",
        "job_id": job_id,
        "status_url": f"/background/status/{job_id}",
        "parameters": {"days_old": days_old},
    }


# --- Audit log endpoint ---

@router.get("/audit/log")
def get_audit_log(
    limit: int = Query(default=20, ge=1, le=100),
    action: Optional[str] = Query(default=None, description="Filter by action (CREATE, UPDATE, DELETE)"),
):
    """Returns the latest entries from the audit log."""
    entries = AUDIT_LOG.copy()
    if action:
        entries = [e for e in entries if e["action"] == action.upper()]

    entries = sorted(entries, key=lambda e: e["timestamp"], reverse=True)
    return {"total": len(entries), "entries": entries[:limit]}

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

from fastapi import FastAPI, HTTPException
from app.routers import tasks
from app.background import job_tracker, BackgroundJobStatus

app = FastAPI(
    title="Task Manager API",
    description="An API with background processing. Module 4 — FastAPI Advanced Features.",
    version="4.0.0",
)

app.include_router(tasks.router)


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


@app.get(
    "/background/status/{job_id}",
    response_model=BackgroundJobStatus,
    tags=["Background"],
    summary="The state of a background operation",
)
def get_background_status(job_id: str):
    """Checks the state of a background operation by job_id."""
    if job_id not in job_tracker:
        raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
    return job_tracker[job_id]


@app.get(
    "/background/jobs",
    response_model=list[BackgroundJobStatus],
    tags=["Background"],
    summary="List every job",
)
def list_background_jobs():
    """Lists every job registered in the tracker."""
    jobs = sorted(
        job_tracker.values(),
        key=lambda j: j.created_at,
        reverse=True,
    )
    return list(jobs)

Step 6: Run and verify

uvicorn app.main:app --reload

Check 1: Create a task with a notification

curl -s -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Review PR #42", "assignee_email": "dev@team.com", "priority": 3}' \
  | python -m json.tool
{
    "id": 9,
    "title": "Review PR #42",
    "status": "pending",
    "priority": 3,
    "assignee_email": "dev@team.com",
    "created_at": "2026-03-13T...",
    "updated_at": null,
    "description": null
}

In the uvicorn console (after the response):

[START] send_notification
[NOTIFICATION] To: dev@team.com | Subject: New task assigned: Review PR #42 | Body: You were assigned task #9 with priority 3.
[DONE] send_notification (1.0s)
[START] write_audit_log
[AUDIT] CREATE task:9 — Title: 'Review PR #42', Priority: 3
[DONE] write_audit_log (0.0s)

Check 2: Create a task with no email — audit log only

curl -s -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Unassigned task", "priority": 2}' | python -m json.tool

Console: just the audit log, no email notification (because there's no assignee_email).

Check 3: Complete a task — the conditional notification

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

Console:

[START] send_notification
[NOTIFICATION] To: devops@team.com | Subject: Task completed: Set up CI/CD pipeline
[DONE] send_notification (1.0s)
[START] send_notification
[NOTIFICATION] To: team@example.com | Subject: Task #1 completed
[DONE] send_notification (1.0s)
[START] write_audit_log
[AUDIT] UPDATE task:1 — Fields updated: ['status']
[DONE] write_audit_log (0.0s)

Two notifications: one to the assignee and one to the team at large. Only because the status changed to "completed".

Check 4: Update without completing — audit only

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

Console: just the audit log, no notification (the status didn't change to "completed").

Check 5: Delete a task

curl -s -X DELETE http://127.0.0.1:8000/tasks/4 | python -m json.tool
{"message": "Task 4 deleted", "deleted_id": 4}

Console: the audit log records the deletion.

Check 6: Cleanup with status tracking

curl -s -X POST "http://127.0.0.1:8000/tasks/cleanup?days_old=30" | python -m json.tool
{
    "message": "Cleanup started",
    "job_id": "a1b2c3d4",
    "status_url": "/background/status/a1b2c3d4",
    "parameters": {"days_old": 30}
}

Immediately:

curl -s http://127.0.0.1:8000/background/status/a1b2c3d4 | python -m json.tool
{
    "job_id": "a1b2c3d4",
    "job_type": "cleanup_completed_tasks",
    "status": "running",
    "result": null,
    "error": null,
    "created_at": "2026-03-13T...",
    "completed_at": null
}

After ~2 seconds:

curl -s http://127.0.0.1:8000/background/status/a1b2c3d4 | python -m json.tool
{
    "job_id": "a1b2c3d4",
    "job_type": "cleanup_completed_tasks",
    "status": "completed",
    "result": "Removed 3 completed tasks older than 30 days",
    "error": null,
    "created_at": "2026-03-13T...",
    "completed_at": "2026-03-13T..."
}

Check 7: Verify the cleanup — fewer tasks

curl -s http://127.0.0.1:8000/tasks | python -m json.tool

The tasks completed more than 30 days ago (IDs 3, 5, 6) no longer show up.

Check 8: The audit log

curl -s http://127.0.0.1:8000/tasks/audit/log | python -m json.tool
{
    "total": 5,
    "entries": [
        {"timestamp": "...", "action": "DELETE", "resource_type": "task", "resource_id": 4, "details": "Deleted task: 'Refactor the auth module'"},
        {"timestamp": "...", "action": "UPDATE", "resource_type": "task", "resource_id": 2, "details": "Fields updated: ['priority']"},
        {"timestamp": "...", "action": "UPDATE", "resource_type": "task", "resource_id": 1, "details": "Fields updated: ['status']"},
        {"timestamp": "...", "action": "CREATE", "resource_type": "task", "resource_id": 10, "details": "..."},
        {"timestamp": "...", "action": "CREATE", "resource_type": "task", "resource_id": 9, "details": "..."}
    ]
}

Check 9: Filter the audit log by action

curl -s "http://127.0.0.1:8000/tasks/audit/log?action=CREATE" | python -m json.tool

CREATE entries only.

Check 10: List the background jobs

curl -s http://127.0.0.1:8000/background/jobs | python -m json.tool

Lists every job registered in the tracker along with its current state.

Completeness checklist

Background Tasks — Notifications:
- [ ] POST /tasks sends a notification if there's an assignee_email
- [ ] POST /tasks with no assignee_email does NOT send a notification
- [ ] PATCH /tasks/{id} with status→completed sends a notification to the assignee
- [ ] PATCH /tasks/{id} with status→completed sends a notification to the team
- [ ] PATCH /tasks/{id} with no change to completed does NOT send a notification
- [ ] The notifications get logged with [NOTIFICATION] in the console

Background Tasks — Audit Log:
- [ ] POST /tasks records a CREATE in the audit log
- [ ] PATCH /tasks/{id} records an UPDATE with the modified fields
- [ ] DELETE /tasks/{id} records a DELETE with the task's title
- [ ] GET /tasks/audit/log returns the entries sorted by timestamp
- [ ] GET /tasks/audit/log accepts a filter by action

Background Tasks — Cleanup:
- [ ] POST /tasks/cleanup accepts a days_old parameter
- [ ] It returns 202 with a job_id and a status_url
- [ ] The cleanup deletes tasks completed > N days ago
- [ ] The job tracker shows the progress (pending → running → completed)
- [ ] GET /background/status/{job_id} returns the current state

Background Tasks — Error Handling:
- [ ] The @safe_task decorator logs the start, the finish, and any errors
- [ ] Background errors do NOT affect the response to the client
- [ ] Every background task has a try/except

Infrastructure:
- [ ] GET /background/jobs lists every registered job
- [ ] GET /background/status/{job_id} returns 404 for a job that doesn't exist
- [ ] The job tracker cleans itself up automatically (cleanup_old_jobs)
- [ ] Logging with the format timestamp - level - message

Troubleshooting

Problem 1: The notifications don't show up in the console

Cause: Logging isn't configured, or you're looking at the wrong terminal.

Fix: Check that logging.basicConfig(level=logging.INFO) is in background.py. The logs show up in the terminal where uvicorn is running, not the one where you run curl.

Problem 2: The audit log is empty after creating tasks

Cause: The audit log is written in the background. If you hit GET /tasks/audit/log immediately after the POST, the background task may not have finished yet.

Fix: Wait a second before checking the audit log, or confirm in the console that the [AUDIT] log appeared.

Problem 3: The cleanup doesn't delete any tasks

Cause: The sample tasks don't meet the condition: they need status=="completed" AND an updated_at earlier than the cutoff. If updated_at is None, they don't get deleted.

Fix: Check that the sample tasks in data.py have updated_at set, with a date older than days_old:

# This task DOES get deleted with days_old=30 (updated_at 30+ days ago)
TaskInDB(
    status="completed",
    updated_at=datetime.now() - timedelta(days=35),
    ...
)

# This task does NOT get deleted (updated_at is None)
TaskInDB(
    status="completed",
    updated_at=None,
    ...
)

Problem 4: GET /tasks/audit/log returns 422 — "path not found"

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

Fix: Declare the fixed routes (/audit/log, /cleanup, /export/csv) BEFORE /{task_id} in the router:

@router.get("/audit/log")     # fixed — first
@router.post("/cleanup")      # fixed — first
@router.get("/{task_id}")     # parameterized — last

Problem 5: The job tracker grows indefinitely

Cause: Every cleanup creates a job entry that never gets deleted.

Fix: The cleanup endpoint already includes cleanup_old_jobs as an extra background task that clears entries older than 60 minutes. If you need more aggressive cleanup, lower max_age_minutes.

Problem 6: Two simultaneous cleanups cause inconsistent data

Cause: If two POST /tasks/cleanup calls arrive at the same time, both iterate over the same tasks_db list and may try to remove the same task twice.

Fix: For this project with in-memory data, that's an acceptable edge case. In production, you'd use a database with transactions. For this project, you can check that the task still exists before removing it:

for task in to_remove:
    if task in tasks_db:
        tasks_db.remove(task)

Problem 7: The task-completed notifications get sent twice

Cause: If the update endpoint gets called twice with status=completed, the first call meets the old_status != "completed" condition, but the second one shouldn't (because old_status is already "completed"). Check that you're capturing old_status BEFORE applying the updates.

Fix:

# ✅ Capture old_status BEFORE updating
old_status = task.status
for field, value in update_data.items():
    setattr(task, field, value)

# Now the comparison is correct
if task.status == "completed" and old_status != "completed":
    # Only runs the first time

Problem 8: The background task logs get interleaved with the request logs

Cause: The background task logs and the uvicorn/FastAPI logs share the same output. That's normal and expected.

Fix: Use a logging format that includes the logger's name so you can tell them apart:

logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

The background logs show up as background - INFO, uvicorn's as uvicorn - INFO.

The patterns you applied

1. Conditional notification

Notifications only go out when they're relevant: when a task is created with an assignee, and when a task is completed (not when the priority changes). That avoids notification spam.

2. A complete audit trail

Every write operation (create, update, delete) leaves an entry in the audit log. The log includes what changed, when, and on which resource. This is standard in production APIs for compliance and debugging.

3. Status tracking for long operations

The cleanup uses the 202 Accepted + polling pattern: the client gets an immediate job_id and can check GET /background/status/{job_id} to see the progress. This is the standard REST pattern for asynchronous operations.

4. A decorator for error handling

@safe_task removes the repeated try/except from every background function. Any new background function just needs the decorator to get automatic logging of its start, its finish, and any errors.

5. A separation of concerns

The background functions live in background.py, separate from the endpoints. The endpoints only decide WHEN to launch tasks; the functions decide WHAT to do. That makes both of them testable independently.

Summary

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

  • Notifications with BackgroundTasks — a simulated email (a log) when tasks are created and completed
  • Conditional notification — only when the status changes to "completed", not on any update
  • A background audit log — every create, update, and delete records an entry
  • Cleanup with status tracking — the 202 + polling pattern for long operations
  • The @safe_task decorator — automatic error handling and logging in every background function
  • A job tracker — an in-memory dict for checking the state of background operations
  • Tracker cleanup — automatic removal of old entries

Your API doesn't just respond anymore — it processes, notifies, logs, and cleans up in the background. The client gets immediate responses while all the secondary processing happens quietly afterward.

Additional resources

  1. FastAPI - Background Tasks — The official BackgroundTasks tutorial
  2. HTTP 202 Accepted — The status code for accepted asynchronous operations
  3. Python logging — The standard library's logging module
  4. Python functools — functools.wraps for decorators
  5. FastAPI - Dependencies — Combining BackgroundTasks with Depends
  6. Audit Logging Best Practices — OWASP guidelines for audit logging

What's next?

Your API responds professionally (M3) and processes in the background (M4). But the communication is still request-response: the client asks, the API answers. There's no way for the API to tell the client something changed unless the client asks.

In Module 5 (WebSockets and File Uploads) you'll add bidirectional real-time communication and file handling:

Module 4 (now):                          Module 5 (next):
──────────────────                        ─────────────────────
Request → Response (one-way)           →  WebSocket (two-way, real-time)
Polling for status                     →  Push notifications to the client
No file handling                       →  File uploads with validation
Background tasks that don't notify     →  Background tasks that notify via WebSocket

Module 4 complete. You've mastered background processing. Next stop: WebSockets and File Uploads, where your API talks to its clients in real time.

← PreviousCelery and RQ — An IntroductionNext →Module 5: WebSockets and File Uploads