Patterns and Error Handling in Background Tasks

Capsule overview

In the previous capsule you learned the basic mechanism of BackgroundTasks: adding functions that run after the response. But the basic mechanism leaves questions unanswered: what happens if a background task fails? Does the client find out? How do you chain tasks that depend on one another? How do you know whether a background task finished or is still running?

In this capsule you'll answer those questions with advanced patterns: robust error handling with try/except and logging, task chaining (one task launching another), tasks that use dependencies, and an in-memory status tracking pattern so you can check the state of your background operations. These patterns are the difference between background tasks that "probably work" and background tasks you can actually monitor and debug.

Error handling: the silent problem

Background tasks run after the client has received their response. If they fail, the client never finds out — they're already gone. Without error handling, failures are silent:

from fastapi import FastAPI, BackgroundTasks
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


def send_email(to: str, subject: str):
    """A function that can fail."""
    if "@" not in to:
        raise ValueError(f"Invalid email: {to}")
    logger.info(f"Email sent to {to}: {subject}")


@app.post("/tasks")
def create_task(title: str, background_tasks: BackgroundTasks):
    task = {"id": 1, "title": title}

    background_tasks.add_task(send_email, "invalid-email", f"Task: {title}")

    return task

curl -X POST "http://127.0.0.1:8000/tasks?title=Test"
# → {"id": 1, "title": "Test"}  ← the client gets a normal response
# In the console: ERROR — ValueError: Invalid email: invalid-email

The client got status 200 and their task. But the email failed silently. In the uvicorn console you see the traceback, but in production that error could go unnoticed if you don't have proper logging.

Pattern: try/except with logging

The fix is to wrap the background task's logic in try/except:

from fastapi import FastAPI, BackgroundTasks
import logging
import time
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


def send_email_safe(to: str, subject: str):
    """A background task with error handling."""
    try:
        logger.info(f"Sending email to {to}...")
        time.sleep(1)

        if "@" not in to:
            raise ValueError(f"Invalid email: {to}")

        logger.info(f"Email sent successfully to {to}: {subject}")

    except ValueError as e:
        logger.error(f"Validation error in email: {e}")

    except Exception as e:
        logger.error(f"Unexpected error sending email to {to}: {type(e).__name__}: {e}")


@app.post("/tasks", status_code=201)
def create_task(title: str, email: str, background_tasks: BackgroundTasks):
    task = {"id": 1, "title": title}

    background_tasks.add_task(send_email_safe, email, f"New task: {title}")

    return task

curl -X POST "http://127.0.0.1:8000/tasks?title=Deploy&email=invalid"
# → {"id": 1, "title": "Deploy"} ← normal response
# Console: ERROR: Validation error in email: Invalid email: invalid

curl -X POST "http://127.0.0.1:8000/tasks?title=Deploy&email=dev@test.com"
# → {"id": 1, "title": "Deploy"} ← normal response
# Console: INFO: Email sent successfully to dev@test.com

The error gets logged but doesn't affect the client. In production, you'd wire these logs into an alerting system (Sentry, CloudWatch, etc.).

Pattern: a generic error handling wrapper

Instead of repeating try/except in every function, create a wrapper:

from fastapi import FastAPI, BackgroundTasks
import logging
import time
import functools

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


def safe_background_task(func):
    """A decorator that wraps background tasks with error handling."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        task_name = func.__name__
        try:
            logger.info(f"[BG-START] {task_name}")
            result = func(*args, **kwargs)
            logger.info(f"[BG-DONE] {task_name}")
            return result
        except Exception as e:
            logger.error(f"[BG-ERROR] {task_name}: {type(e).__name__}: {e}")
    return wrapper


@safe_background_task
def send_email(to: str, subject: str):
    time.sleep(1)
    logger.info(f"Email sent to {to}: {subject}")


@safe_background_task
def update_analytics(event: str, data: dict):
    logger.info(f"Analytics updated: {event} — {data}")


@safe_background_task
def cleanup_temp_data(older_than_days: int):
    logger.info(f"Cleaned up data older than {older_than_days} days")


@app.post("/tasks", status_code=201)
def create_task(title: str, background_tasks: BackgroundTasks):
    task = {"id": 1, "title": title}

    background_tasks.add_task(send_email, "team@test.com", f"New task: {title}")
    background_tasks.add_task(update_analytics, "task_created", {"title": title})
    background_tasks.add_task(cleanup_temp_data, 7)

    return task

Every background task now gets automatic logging of its start, its finish, and any error. You don't need to repeat try/except in each function.

Pattern: task chaining

Sometimes a task has to launch another task once it completes. You can't use BackgroundTasks inside a background task directly, but you can chain the logic within the same function:

from fastapi import FastAPI, BackgroundTasks
import logging
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


def process_order_chain(order_id: int, customer_email: str):
    """A chain of post-order operations."""

    logger.info(f"[Step 1] Validating inventory for order #{order_id}...")
    time.sleep(1)
    logger.info(f"[Step 1] Inventory validated ✓")

    logger.info(f"[Step 2] Processing payment for order #{order_id}...")
    time.sleep(1)
    logger.info(f"[Step 2] Payment processed ✓")

    logger.info(f"[Step 3] Sending confirmation to {customer_email}...")
    time.sleep(1)
    logger.info(f"[Step 3] Confirmation sent ✓")

    logger.info(f"[COMPLETE] Order #{order_id} processed successfully")


@app.post("/orders", status_code=201)
def create_order(product: str, email: str, background_tasks: BackgroundTasks):
    order = {"id": 42, "product": product, "status": "processing"}

    background_tasks.add_task(process_order_chain, 42, email)

    return order

The process_order_chain function runs three steps sequentially. If any step fails, you can wrap each one in try/except and handle the partial failure:

def process_order_chain_safe(order_id: int, customer_email: str):
    """A chain with error handling per step."""
    steps_completed = []

    try:
        logger.info(f"[Step 1] Validating inventory...")
        time.sleep(1)
        steps_completed.append("inventory_validated")

        logger.info(f"[Step 2] Processing payment...")
        time.sleep(1)
        steps_completed.append("payment_processed")

        logger.info(f"[Step 3] Sending confirmation...")
        time.sleep(1)
        steps_completed.append("confirmation_sent")

    except Exception as e:
        logger.error(
            f"Order #{order_id} failed at step {len(steps_completed) + 1}: "
            f"{type(e).__name__}: {e}. "
            f"Completed steps: {steps_completed}"
        )

Pattern: in-memory status tracking

The most useful pattern for background tasks: tracking their state so the client can check on them:

from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from datetime import datetime
from typing import Optional, Literal
import logging
import time
import uuid

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


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


task_tracker: dict[str, TaskStatus] = {}


def run_tracked_task(task_id: str, operation: str, duration: int):
    """A background task with status tracking."""
    task_tracker[task_id].status = "running"
    logger.info(f"[{task_id}] Starting: {operation}")

    try:
        time.sleep(duration)

        task_tracker[task_id].status = "completed"
        task_tracker[task_id].result = f"{operation} completed successfully"
        task_tracker[task_id].completed_at = datetime.now()
        logger.info(f"[{task_id}] Completed: {operation}")

    except Exception as e:
        task_tracker[task_id].status = "failed"
        task_tracker[task_id].error = str(e)
        task_tracker[task_id].completed_at = datetime.now()
        logger.error(f"[{task_id}] Failed: {operation} — {e}")


@app.post("/reports/generate", status_code=202)
def generate_report(report_type: str, background_tasks: BackgroundTasks):
    """Kicks off report generation in the background. Returns an ID to check the status."""
    task_id = str(uuid.uuid4())[:8]

    task_tracker[task_id] = TaskStatus(
        task_id=task_id,
        status="pending",
        created_at=datetime.now(),
    )

    background_tasks.add_task(run_tracked_task, task_id, f"Generate {report_type} report", 5)

    return {
        "message": "Report generation started",
        "task_id": task_id,
        "status_url": f"/reports/status/{task_id}",
    }


@app.get("/reports/status/{task_id}", response_model=TaskStatus)
def get_report_status(task_id: str):
    """Checks the state of a background operation."""
    if task_id not in task_tracker:
        raise HTTPException(status_code=404, detail="Task not found")
    return task_tracker[task_id]

# Step 1: Start the generation
curl -X POST "http://127.0.0.1:8000/reports/generate?report_type=monthly"
# → {"message": "Report generation started", "task_id": "a1b2c3d4", "status_url": "/reports/status/a1b2c3d4"}

# Step 2: Check the status (immediately)
curl http://127.0.0.1:8000/reports/status/a1b2c3d4
# → {"task_id": "a1b2c3d4", "status": "running", "result": null, ...}

# Step 3: Check the status (5 seconds later)
curl http://127.0.0.1:8000/reports/status/a1b2c3d4
# → {"task_id": "a1b2c3d4", "status": "completed", "result": "Generate monthly report completed successfully", ...}

The client:

1. Sends POST /reports/generate → gets back 202 Accepted with a task_id
2. Polls GET /reports/status/{task_id} to check the progress
3. When status == "completed", reads the result

This is the standard pattern for asynchronous operations in REST APIs.

Pattern: long-running vs short tasks

Not all background tasks are the same:

from fastapi import FastAPI, BackgroundTasks
import logging
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


def short_task():
    """< 1 second — needs no tracking."""
    logger.info("Cache invalidated")


def medium_task():
    """1-30 seconds — tracking optional."""
    time.sleep(5)
    logger.info("PDF generated")


def long_task(task_id: str, tracker: dict):
    """> 30 seconds — tracking mandatory."""
    tracker[task_id] = "running"
    for i in range(10):
        time.sleep(3)
        tracker[task_id] = f"step {i + 1}/10"
    tracker[task_id] = "completed"

For short tasks, fire-and-forget is enough. For long tasks, always implement status tracking — the client needs to know the operation is still alive.

Pattern: manual retries

BackgroundTasks has no automatic retry. If you need to retry, do it by hand:

from fastapi import FastAPI, BackgroundTasks
import logging
import time
import random

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


def send_with_retry(to: str, message: str, max_retries: int = 3):
    """Sends a notification with retries."""
    for attempt in range(1, max_retries + 1):
        try:
            logger.info(f"Attempt {attempt}/{max_retries}: Sending to {to}")

            if random.random() < 0.5:
                raise ConnectionError("Service unavailable")

            logger.info(f"Sent successfully to {to}: {message}")
            return

        except ConnectionError as e:
            logger.warning(f"Attempt {attempt} failed: {e}")
            if attempt < max_retries:
                wait = attempt * 2
                logger.info(f"Waiting {wait}s before retry...")
                time.sleep(wait)

    logger.error(f"All {max_retries} attempts failed for {to}")


@app.post("/notifications")
def send_notification(to: str, message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_with_retry, to, message)
    return {"message": "Notification queued", "to": to}

Exponential backoff: the first retry waits 2 seconds, the second 4, the third 6. This keeps you from hammering a service that's already struggling.

Connection with the project

In the project capsule (05), you'll combine these patterns: error handling with the safe_background_task wrapper, status tracking for long operations like cleanup, and retries for notifications. Together they produce a background processing system that's robust and observable.

Troubleshooting

Problem 1: The status tracker shows "pending" forever

Cause: The background task failed with an uncaught exception and never updated the status to "failed".

Fix: Wrap the background task in a try/except that always updates the status:

def tracked_task(task_id, tracker):
    try:
        tracker[task_id] = "running"
        # ... work ...
        tracker[task_id] = "completed"
    except Exception as e:
        tracker[task_id] = "failed"  # Always update on error
        logger.error(f"Task {task_id} failed: {e}")

Problem 2: The task_tracker grows forever

Cause: You never clean up the completed entries. In production, the dict grows with every request.

Fix: Add periodic cleanup or a TTL:

from datetime import timedelta

def cleanup_tracker(tracker: dict, max_age_minutes: int = 30):
    cutoff = datetime.now() - timedelta(minutes=max_age_minutes)
    expired = [
        tid for tid, status in tracker.items()
        if status.completed_at and status.completed_at < cutoff
    ]
    for tid in expired:
        del tracker[tid]

Problem 3: Chained tasks fail silently in the middle

Cause: If step 2 of a chain fails and you have no logging, you have no idea where it stopped.

Fix: Log each step with its number and record which steps completed:

def chain_task(order_id):
    completed = []
    try:
        step1()
        completed.append("step1")
        step2()
        completed.append("step2")
    except Exception as e:
        logger.error(f"Chain failed. Completed: {completed}. Error: {e}")

Problem 4: Retries with exponential backoff cause timeouts in tests

Cause: time.sleep(4) in a retry makes the tests take forever.

Fix: Make the delays configurable:

def send_with_retry(to, message, max_retries=3, base_delay=2):
    for attempt in range(1, max_retries + 1):
        try:
            # ...
            return
        except Exception:
            if attempt < max_retries:
                time.sleep(base_delay * attempt)

# In tests: send_with_retry(to, msg, base_delay=0)

Problem 5: Two background tasks modify the same data at the same time

Cause: A race condition. If two requests arrive simultaneously and both launch background tasks that modify the same list, you can get data corruption.

Fix: For in-memory data, use a lock:

import threading

task_lock = threading.Lock()

def safe_update(data_list, item):
    with task_lock:
        data_list.append(item)

In production, use a database or Redis instead of in-memory data.

Exercises

Exercise 1: try/except with logging (Easy)

Create a POST /emails endpoint that accepts to and subject. In the background, try to send the "email". If to doesn't contain @, the function should log an error (not raise an exception). If it's valid, log the success. The endpoint always returns 202.

from fastapi import FastAPI, BackgroundTasks
import logging
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


def send_email(to: str, subject: str):
    try:
        time.sleep(1)
        if "@" not in to:
            raise ValueError(f"Invalid email: {to}")
        logger.info(f"[SUCCESS] Email sent to {to}: {subject}")
    except ValueError as e:
        logger.error(f"[FAILED] {e}")
    except Exception as e:
        logger.error(f"[UNEXPECTED] {type(e).__name__}: {e}")


@app.post("/emails", status_code=202)
def queue_email(to: str, subject: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, to, subject)
    return {"message": "Email queued", "to": to}

curl -X POST "http://127.0.0.1:8000/emails?to=invalid&subject=Test"
# → {"message": "Email queued", "to": "invalid"} (202)
# Console: [FAILED] Invalid email: invalid

curl -X POST "http://127.0.0.1:8000/emails?to=ana@test.com&subject=Test"
# → {"message": "Email queued", "to": "ana@test.com"} (202)
# Console: [SUCCESS] Email sent to ana@test.com: Test

Exercise 2: Basic status tracking (Easy)

Create a POST /jobs endpoint that starts a "job" in the background (simulate it with a 3s sleep). Return a job_id. Create GET /jobs/{job_id}/status that returns the job's current state (pending, running, completed). Use a global dict as the tracker.

from fastapi import FastAPI, BackgroundTasks, HTTPException
from datetime import datetime
import logging
import time
import uuid

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

job_tracker: dict[str, dict] = {}


def run_job(job_id: str, job_type: str):
    job_tracker[job_id]["status"] = "running"
    logger.info(f"Job {job_id} started: {job_type}")

    try:
        time.sleep(3)
        job_tracker[job_id]["status"] = "completed"
        job_tracker[job_id]["completed_at"] = datetime.now().isoformat()
        logger.info(f"Job {job_id} completed")
    except Exception as e:
        job_tracker[job_id]["status"] = "failed"
        job_tracker[job_id]["error"] = str(e)
        logger.error(f"Job {job_id} failed: {e}")


@app.post("/jobs", status_code=202)
def create_job(job_type: str, background_tasks: BackgroundTasks):
    job_id = str(uuid.uuid4())[:8]

    job_tracker[job_id] = {
        "job_id": job_id,
        "job_type": job_type,
        "status": "pending",
        "created_at": datetime.now().isoformat(),
        "completed_at": None,
        "error": None,
    }

    background_tasks.add_task(run_job, job_id, job_type)

    return {"job_id": job_id, "status_url": f"/jobs/{job_id}/status"}


@app.get("/jobs/{job_id}/status")
def get_job_status(job_id: str):
    if job_id not in job_tracker:
        raise HTTPException(status_code=404, detail="Job not found")
    return job_tracker[job_id]

Exercise 3: The safe_background_task decorator (Medium)

Create a @safe_background_task decorator that wraps any function with: (1) a start log with a timestamp, (2) a try/except that logs errors, (3) a finish log with the duration in seconds. Apply it to two different functions and use them as background tasks.

from fastapi import FastAPI, BackgroundTasks
import logging
import time
import functools

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


def safe_background_task(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        name = func.__name__
        start = time.time()
        logger.info(f"[BG-START] {name} at {time.strftime('%H:%M:%S')}")
        try:
            result = func(*args, **kwargs)
            elapsed = round(time.time() - start, 2)
            logger.info(f"[BG-DONE] {name} in {elapsed}s")
            return result
        except Exception as e:
            elapsed = round(time.time() - start, 2)
            logger.error(f"[BG-ERROR] {name} after {elapsed}s: {type(e).__name__}: {e}")
    return wrapper


@safe_background_task
def process_payment(order_id: int, amount: float):
    time.sleep(2)
    logger.info(f"Payment of ${amount} processed for order #{order_id}")


@safe_background_task
def send_receipt(email: str, order_id: int):
    time.sleep(1)
    logger.info(f"Receipt sent to {email} for order #{order_id}")


@app.post("/orders", status_code=201)
def create_order(amount: float, email: str, background_tasks: BackgroundTasks):
    order = {"id": 1, "amount": amount, "status": "created"}

    background_tasks.add_task(process_payment, 1, amount)
    background_tasks.add_task(send_receipt, email, 1)

    return order

Exercise 4: A task chain with error recovery (Medium)

Create a process_pipeline function that runs 4 sequential steps in the background. If a step fails, log which steps completed and which one failed. Simulate step 3 failing at random (50% of the time). Include a tracker_dict parameter to record the progress.

from fastapi import FastAPI, BackgroundTasks, HTTPException
import logging
import time
import random
import uuid

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

pipeline_tracker: dict[str, dict] = {}


def process_pipeline(pipeline_id: str, tracker: dict):
    steps = ["validate_input", "transform_data", "send_to_service", "update_cache"]
    completed = []

    tracker[pipeline_id]["status"] = "running"

    for i, step in enumerate(steps):
        try:
            logger.info(f"[{pipeline_id}] Step {i + 1}/{len(steps)}: {step}")
            time.sleep(1)

            if step == "send_to_service" and random.random() < 0.5:
                raise ConnectionError("External service unavailable")

            completed.append(step)
            tracker[pipeline_id]["progress"] = f"{len(completed)}/{len(steps)}"

        except Exception as e:
            tracker[pipeline_id]["status"] = "failed"
            tracker[pipeline_id]["error"] = f"Failed at '{step}': {e}"
            tracker[pipeline_id]["completed_steps"] = completed
            logger.error(f"[{pipeline_id}] Pipeline failed at '{step}': {e}. Completed: {completed}")
            return

    tracker[pipeline_id]["status"] = "completed"
    tracker[pipeline_id]["completed_steps"] = completed
    logger.info(f"[{pipeline_id}] Pipeline completed successfully")


@app.post("/pipeline/start", status_code=202)
def start_pipeline(background_tasks: BackgroundTasks):
    pipeline_id = str(uuid.uuid4())[:8]

    pipeline_tracker[pipeline_id] = {
        "id": pipeline_id,
        "status": "pending",
        "progress": "0/4",
        "completed_steps": [],
        "error": None,
    }

    background_tasks.add_task(process_pipeline, pipeline_id, pipeline_tracker)

    return {"pipeline_id": pipeline_id, "status_url": f"/pipeline/{pipeline_id}"}


@app.get("/pipeline/{pipeline_id}")
def get_pipeline_status(pipeline_id: str):
    if pipeline_id not in pipeline_tracker:
        raise HTTPException(status_code=404, detail="Pipeline not found")
    return pipeline_tracker[pipeline_id]

Exercise 5: Retries with exponential backoff (Hard)

Create a fetch_with_retry function that simulates calling an external API. The call fails 70% of the time with a ConnectionError. Implement retries with exponential backoff (1s, 2s, 4s) and a maximum of 4 attempts. Log each attempt and the final result. Use it as a background task from a POST /sync endpoint.

from fastapi import FastAPI, BackgroundTasks
import logging
import time
import random

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


def fetch_with_retry(url: str, max_retries: int = 4, base_delay: float = 1.0):
    """Fetch with retries and exponential backoff."""
    for attempt in range(1, max_retries + 1):
        try:
            logger.info(f"[Attempt {attempt}/{max_retries}] Fetching {url}...")
            time.sleep(0.5)

            if random.random() < 0.7:
                raise ConnectionError("Connection refused")

            logger.info(f"[SUCCESS] Fetched {url} on attempt {attempt}")
            return {"url": url, "status": "success", "attempts": attempt}

        except ConnectionError as e:
            logger.warning(f"[Attempt {attempt}] Failed: {e}")

            if attempt < max_retries:
                delay = base_delay * (2 ** (attempt - 1))
                logger.info(f"Waiting {delay}s before retry...")
                time.sleep(delay)

    logger.error(f"[EXHAUSTED] All {max_retries} attempts failed for {url}")
    return {"url": url, "status": "failed", "attempts": max_retries}


@app.post("/sync", status_code=202)
def sync_external_data(url: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(fetch_with_retry, url)
    return {"message": "Sync started", "url": url}

curl -X POST "http://127.0.0.1:8000/sync?url=https://api.example.com/data"
# → {"message": "Sync started", "url": "..."}
# The console shows the attempts with growing delays

Summary

• Error handling is mandatory in background tasks — without try/except, failures are silent
• The @safe_background_task decorator removes the repeated try/except from every function
• Task chaining happens inside a single function that runs the steps sequentially
• Status tracking with an in-memory dict lets you check the state of long operations
• The 202 Accepted pattern + polling with GET /status/{id} is the REST standard for asynchronous operations
• Manual retries with exponential backoff (delay × 2^attempt) is the pattern for unstable services
• Short tasks (< 1s): fire-and-forget. Long tasks (> 30s): status tracking is mandatory
• Race conditions: use locks (threading.Lock) if multiple background tasks modify the same data
• The task_tracker grows: implement periodic cleanup of completed entries

Additional resources

1. FastAPI - Background Tasks — The official tutorial
2. HTTP 202 Accepted — The status code for asynchronous operations
3. Exponential Backoff — The backoff algorithm for retries
4. Python functools.wraps — Preserving metadata in decorators
5. Python threading.Lock — Locks for concurrency
6. Real Python - Decorators — A complete guide to decorators in Python

Next capsule: Celery and RQ — an introduction — When BackgroundTasks isn't enough and how to scale up to distributed task queues.