BackgroundTasks in FastAPI

Capsule overview

FastAPI ships with a built-in mechanism for running functions after the response has been sent to the client: BackgroundTasks. It's an object you inject as a parameter in your endpoint, you add functions to it with add_task(), and FastAPI runs them automatically once the response has gone out. The client gets their response immediately — the background functions run afterward, without blocking anything.

In this capsule you'll learn the basic use of BackgroundTasks: how to inject it, how to add tasks with arguments, how to add multiple tasks to a single endpoint, and the most common patterns — email simulation, structured logging, and data cleanup. All of it with runnable code you can execute and watch in the server console.

Your first background task

The basic pattern has three steps: define the function, inject BackgroundTasks, and add the task:

from fastapi import FastAPI, BackgroundTasks
import logging
import time

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

app = FastAPI()


def send_notification(email: str, message: str):
    """Simulates sending an email (takes 2 seconds)."""
    logger.info(f"Sending email to {email}...")
    time.sleep(2)
    logger.info(f"Email sent to {email}: {message}")


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

    background_tasks.add_task(
        send_notification,
        "team@example.com",
        f"New task created: {title}",
    )

    return task

curl -X POST "http://127.0.0.1:8000/tasks?title=Deploy+v3"

Here's what happens:

1. Immediately: the client receives {"id": 1, "title": "Deploy v3", "status": "pending"}
2. 2 seconds later: in the server console you see:

   INFO: Sending email to team@example.com...
   INFO: Email sent to team@example.com: New task created: Deploy v3
   

The client doesn't wait the 2 seconds the email takes. They get their response instantly.

The anatomy of add_task

background_tasks.add_task(function, arg1, arg2, kwarg1=value1)

The function can be sync or async:

import asyncio

def sync_task(data: str):
    """A sync function — runs in a thread pool."""
    time.sleep(1)
    logger.info(f"Sync task: {data}")


async def async_task(data: str):
    """An async function — runs in the event loop."""
    await asyncio.sleep(1)
    logger.info(f"Async task: {data}")


@app.post("/example")
def example(background_tasks: BackgroundTasks):
    background_tasks.add_task(sync_task, "sync data")
    background_tasks.add_task(async_task, "async data")
    return {"message": "Tasks added"}

Multiple background tasks

You can add several tasks to a single endpoint. They run sequentially, in the order you added them:

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 log_event(event: str, details: str):
    """Records an event in the log."""
    logger.info(f"[EVENT] {event}: {details}")


def send_email(to: str, subject: str):
    """Simulates sending an email."""
    time.sleep(1)
    logger.info(f"[EMAIL] To: {to} | Subject: {subject}")


def update_analytics(action: str, resource_id: int):
    """Simulates updating analytics."""
    time.sleep(0.5)
    logger.info(f"[ANALYTICS] {action} on resource {resource_id}")


@app.post("/tasks", status_code=201)
def create_task(title: str, background_tasks: BackgroundTasks):
    task = {"id": 42, "title": title, "status": "pending", "created_at": datetime.now().isoformat()}

    background_tasks.add_task(log_event, "TASK_CREATED", f"Task '{title}' created")
    background_tasks.add_task(send_email, "team@example.com", f"New task: {title}")
    background_tasks.add_task(update_analytics, "create", 42)

    return task

After the client gets the response, in the console you see (with the delays):

INFO: [EVENT] TASK_CREATED: Task 'Deploy v3' created
INFO: [EMAIL] To: team@example.com | Subject: New task: Deploy v3
INFO: [ANALYTICS] create on resource 42

The three functions run in order: log → email → analytics.

BackgroundTasks with Depends

BackgroundTasks integrates with dependency injection. You can receive it inside a dependency:

from fastapi import FastAPI, BackgroundTasks, Depends
import logging

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

app = FastAPI()


def audit_logger(background_tasks: BackgroundTasks):
    """A dependency that adds automatic logging."""
    def log_action(action: str, resource_type: str, resource_id: int):
        logger.info(f"[AUDIT] {action} {resource_type}:{resource_id}")

    def add_audit(action: str, resource_type: str, resource_id: int):
        background_tasks.add_task(log_action, action, resource_type, resource_id)

    return add_audit


@app.post("/tasks", status_code=201)
def create_task(title: str, audit: callable = Depends(audit_logger)):
    task = {"id": 1, "title": title}
    audit("CREATE", "task", 1)
    return task


@app.delete("/tasks/{task_id}")
def delete_task(task_id: int, audit: callable = Depends(audit_logger)):
    audit("DELETE", "task", task_id)
    return {"message": f"Task {task_id} deleted"}

The audit_logger dependency receives BackgroundTasks automatically (FastAPI injects it) and returns an add_audit function that adds logs in the background. Every endpoint that uses this dependency gets automatic logging with no repetitive code.

Pattern: email notification (simulated)

The most common use case. It simulates sending an email with logging:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field
from datetime import datetime
import logging
import time

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

app = FastAPI()


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    assignee_email: str
    priority: int = Field(default=1, ge=1, le=5)


def send_task_notification(email: str, task_title: str, task_id: int):
    """Simulates sending an email notification."""
    logger.info(f"Preparing email for {email}...")
    time.sleep(2)
    logger.info(
        f"Email sent to {email}: "
        f"'You were assigned task #{task_id}: {task_title}'"
    )


tasks_db: list[dict] = []


@app.post("/tasks", status_code=201)
def create_task(task: TaskCreate, background_tasks: BackgroundTasks):
    new_id = len(tasks_db) + 1
    task_data = {
        "id": new_id,
        "title": task.title,
        "assignee_email": task.assignee_email,
        "priority": task.priority,
        "status": "pending",
        "created_at": datetime.now().isoformat(),
    }
    tasks_db.append(task_data)

    background_tasks.add_task(
        send_task_notification,
        task.assignee_email,
        task.title,
        new_id,
    )

    return task_data

curl -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Review PR #42", "assignee_email": "dev@example.com", "priority": 3}'

The response arrives immediately. Two seconds later, the log confirms the "email" went out.

Pattern: structured logging

Record operations with context so you can debug them:

from fastapi import FastAPI, BackgroundTasks, Request
import logging
import json
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("api.audit")

app = FastAPI()


def write_audit_log(
    action: str,
    resource: str,
    resource_id: int,
    user_agent: str,
    timestamp: str,
):
    """Writes a structured audit log entry."""
    log_entry = {
        "action": action,
        "resource": resource,
        "resource_id": resource_id,
        "user_agent": user_agent,
        "timestamp": timestamp,
    }
    logger.info(f"AUDIT: {json.dumps(log_entry)}")


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

    background_tasks.add_task(
        write_audit_log,
        action="CREATE",
        resource="task",
        resource_id=1,
        user_agent=request.headers.get("user-agent", "unknown"),
        timestamp=datetime.now().isoformat(),
    )

    return task

Pattern: data cleanup

Clean up temporary or expired data after an operation:

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

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

app = FastAPI()

temp_files: list[dict] = [
    {"name": "report_old.csv", "created": datetime.now() - timedelta(days=10)},
    {"name": "export_recent.csv", "created": datetime.now() - timedelta(hours=2)},
    {"name": "backup_old.zip", "created": datetime.now() - timedelta(days=30)},
]


def cleanup_old_files(max_age_days: int = 7):
    """Deletes temporary files older than max_age_days."""
    cutoff = datetime.now() - timedelta(days=max_age_days)
    old_files = [f for f in temp_files if f["created"] < cutoff]

    for f in old_files:
        temp_files.remove(f)
        logger.info(f"Deleted old file: {f['name']}")

    logger.info(f"Cleanup complete: {len(old_files)} files removed, {len(temp_files)} remaining")


@app.post("/reports/generate")
def generate_report(background_tasks: BackgroundTasks):
    report = {"name": f"report_{datetime.now().strftime('%Y%m%d')}.csv", "created": datetime.now()}
    temp_files.append(report)

    background_tasks.add_task(cleanup_old_files, max_age_days=7)

    return {"message": "Report generated", "filename": report["name"]}

Every time a new report gets generated, the old files get cleaned up in the background.

BackgroundTasks vs doing everything in the endpoint

To see the difference in timing:

from fastapi import FastAPI, BackgroundTasks
import time
import logging

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

app = FastAPI()


def slow_operation(label: str):
    time.sleep(3)
    logger.info(f"Slow operation '{label}' completed")


@app.post("/without-background")
def without_background():
    """Everything synchronous — the client waits."""
    start = time.time()
    slow_operation("inline")
    elapsed = time.time() - start
    return {"message": "Done", "response_time_seconds": round(elapsed, 2)}


@app.post("/with-background")
def with_background(background_tasks: BackgroundTasks):
    """The slow operation in the background — the client doesn't wait."""
    start = time.time()
    background_tasks.add_task(slow_operation, "background")
    elapsed = time.time() - start
    return {"message": "Done", "response_time_seconds": round(elapsed, 2)}

curl -X POST http://127.0.0.1:8000/without-background
# → {"message": "Done", "response_time_seconds": 3.0}  ← 3 seconds of waiting

curl -X POST http://127.0.0.1:8000/with-background
# → {"message": "Done", "response_time_seconds": 0.0}  ← immediate
# (3 seconds later, in the console: "Slow operation 'background' completed")

Connection with the project

In the project capsule (05), you'll use BackgroundTasks for three operations: an email notification when tasks are created, audit logging when tasks are modified/deleted, and cleanup of completed tasks. Each one runs after the response without affecting the client's experience.

Troubleshooting

Problem 1: The background task doesn't run

Cause: You're probably not injecting BackgroundTasks as a parameter of the endpoint.

Fix:

# ❌ BackgroundTasks isn't injected — the variable doesn't exist
@app.post("/tasks")
def create_task(title: str):
    background_tasks.add_task(...)  # NameError

# ✅ Inject it as a parameter
@app.post("/tasks")
def create_task(title: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(...)

Problem 2: I don't see the background task's logs

Cause: Logging isn't configured, or the level is too high.

Fix:

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

# Make sure you're using logger.info(), not print()
# print() may not show up immediately because of buffering

Problem 3: The background task gets the wrong arguments

Cause: add_task passes positional arguments. If you mix up the order, the values arrive wrong.

Fix: Use keyword arguments for clarity:

# ❌ Confusing positionals
background_tasks.add_task(send_email, "subject", "to@email.com")

# ✅ Explicit keywords
background_tasks.add_task(send_email, to="to@email.com", subject="subject")

Problem 4: The server restarts and the tasks are lost

Cause: This is the expected behavior. BackgroundTasks is in-process — if uvicorn restarts (because of --reload or a crash), the pending tasks are lost.

Fix: For tasks that can't be lost, use Celery or RQ (Capsule 04). For development with --reload, this is normal and acceptable.

Problem 5: A background task blocks other requests

Cause: A heavy sync function in the background task can block if the thread pool is full.

Fix: For very heavy work, use an async function or consider Celery:

# ❌ A sync function that blocks for a long time
def heavy_sync_task():
    time.sleep(60)  # Blocks a pool thread for 60 seconds

# ✅ An async function that doesn't block the thread pool
async def heavy_async_task():
    await asyncio.sleep(60)  # Blocks no threads

Exercises

Exercise 1: A basic background task with logging (Easy)

Create a POST /orders endpoint that accepts product and customer_email. Return the created order immediately, and in the background log "Processing order of {product} for {customer_email}" with a 1-second delay.

from fastapi import FastAPI, BackgroundTasks
import logging
import time

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

app = FastAPI()


def process_order(product: str, email: str):
    time.sleep(1)
    logger.info(f"Processing order of {product} for {email}")


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

    background_tasks.add_task(process_order, product, customer_email)

    return order

curl -X POST "http://127.0.0.1:8000/orders?product=Laptop&customer_email=ana@test.com"
# → Immediate response
# (1 second later, in the console): INFO: Processing order of Laptop for ana@test.com

Exercise 2: Multiple background tasks (Easy)

Create a POST /users/register endpoint that accepts name and email. After returning the response, run three background tasks: (1) log the registration, (2) send a welcome email (2s delay), (3) add the user to the newsletter list. Each one should log a different message.

from fastapi import FastAPI, BackgroundTasks
import logging
import time

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

app = FastAPI()


def log_registration(name: str):
    logger.info(f"[REGISTER] New user registered: {name}")


def send_welcome_email(email: str, name: str):
    time.sleep(2)
    logger.info(f"[EMAIL] Welcome email sent to {email} for {name}")


def add_to_newsletter(email: str):
    logger.info(f"[NEWSLETTER] {email} added to newsletter list")


@app.post("/users/register", status_code=201)
def register_user(name: str, email: str, background_tasks: BackgroundTasks):
    user = {"id": 1, "name": name, "email": email, "active": True}

    background_tasks.add_task(log_registration, name)
    background_tasks.add_task(send_welcome_email, email, name)
    background_tasks.add_task(add_to_newsletter, email)

    return user

curl -X POST "http://127.0.0.1:8000/users/register?name=Ana&email=ana@test.com"
# → Immediate response
# Console (in order):
#   [REGISTER] New user registered: Ana
#   [EMAIL] Welcome email sent to ana@test.com for Ana (2s later)
#   [NEWSLETTER] ana@test.com added to newsletter list

Exercise 3: A background task with Depends (Medium)

Create a notification_service dependency that receives BackgroundTasks and returns a notify(message, channel) function. The function logs the message along with the channel. Use this dependency in two endpoints: POST /tasks and DELETE /tasks/{id}.

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

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

app = FastAPI()


def send_notification(message: str, channel: str):
    time.sleep(0.5)
    logger.info(f"[{channel.upper()}] {message}")


def notification_service(background_tasks: BackgroundTasks):
    def notify(message: str, channel: str = "general"):
        background_tasks.add_task(send_notification, message, channel)
    return notify


@app.post("/tasks", status_code=201)
def create_task(title: str, notify=Depends(notification_service)):
    task = {"id": 1, "title": title, "status": "pending"}
    notify(f"Task created: {title}", "tasks")
    return task


@app.delete("/tasks/{task_id}")
def delete_task(task_id: int, notify=Depends(notification_service)):
    notify(f"Task #{task_id} deleted", "tasks")
    notify(f"Cleanup needed for task #{task_id}", "maintenance")
    return {"message": f"Task {task_id} deleted"}

curl -X POST "http://127.0.0.1:8000/tasks?title=Deploy"
# → Immediate response + [TASKS] Task created: Deploy

curl -X DELETE http://127.0.0.1:8000/tasks/1
# → Immediate response + [TASKS] Task #1 deleted + [MAINTENANCE] Cleanup needed...

Exercise 4: Conditional — a background task only if a condition is met (Medium)

Create a PATCH /tasks/{id} endpoint that updates the status. If the new status is "completed", add a background task that logs "Task #{id} completed — notifying the team". If the status isn't "completed", don't add any background task.

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

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

app = FastAPI()

tasks_db = [
    {"id": 1, "title": "Setup CI/CD", "status": "in_progress"},
    {"id": 2, "title": "Write docs", "status": "pending"},
]


class TaskUpdate(BaseModel):
    status: Optional[Literal["pending", "in_progress", "completed"]] = None
    title: Optional[str] = None


def notify_completion(task_id: int, task_title: str):
    time.sleep(1)
    logger.info(f"Task #{task_id} '{task_title}' completed — notifying the team")


@app.patch("/tasks/{task_id}")
def update_task(task_id: int, updates: TaskUpdate, background_tasks: BackgroundTasks):
    task = next((t for t in tasks_db if t["id"] == task_id), None)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")

    update_data = updates.model_dump(exclude_unset=True)
    was_completed = task["status"] == "completed"

    for field, value in update_data.items():
        task[field] = value

    if task["status"] == "completed" and not was_completed:
        background_tasks.add_task(notify_completion, task_id, task["title"])

    return task

curl -X PATCH http://127.0.0.1:8000/tasks/1 \
  -H "Content-Type: application/json" \
  -d '{"status": "completed"}'
# → Immediate response
# 1 second later: Task #1 'Setup CI/CD' completed — notifying the team

curl -X PATCH http://127.0.0.1:8000/tasks/2 \
  -H "Content-Type: application/json" \
  -d '{"status": "in_progress"}'
# → Immediate response, no background task (it isn't "completed")

Exercise 5: A background task that writes to a file (Hard)

Create a POST /events endpoint that accepts an event_type and details. Return confirmation immediately. In the background, write the event to an events.log file with a timestamp, the type, and the details. Each event is a new line. Check that the file gets updated after every request.

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from datetime import datetime
from pathlib import Path
import logging

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

app = FastAPI()

LOG_FILE = Path("events.log")


class EventCreate(BaseModel):
    event_type: str
    details: str


def write_event_to_file(event_type: str, details: str, timestamp: str):
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(f"[{timestamp}] {event_type}: {details}\n")
    logger.info(f"Event written to {LOG_FILE}: {event_type}")


@app.post("/events", status_code=201)
def create_event(event: EventCreate, background_tasks: BackgroundTasks):
    timestamp = datetime.now().isoformat()

    background_tasks.add_task(
        write_event_to_file,
        event.event_type,
        event.details,
        timestamp,
    )

    return {
        "message": "Event registered",
        "event_type": event.event_type,
        "timestamp": timestamp,
    }


@app.get("/events/log")
def read_events_log():
    if not LOG_FILE.exists():
        return {"events": []}
    content = LOG_FILE.read_text(encoding="utf-8")
    return {"events": content.strip().split("\n") if content.strip() else []}

curl -X POST http://127.0.0.1:8000/events \
  -H "Content-Type: application/json" \
  -d '{"event_type": "USER_LOGIN", "details": "User ana@test.com logged in"}'
# → Immediate response

curl http://127.0.0.1:8000/events/log
# → {"events": ["[2026-03-13T...] USER_LOGIN: User ana@test.com logged in"]}

Summary

• BackgroundTasks is injected as a parameter of the endpoint — FastAPI provides it automatically
• add_task(func, *args, **kwargs) registers a function to run after the response
• Multiple tasks get added with multiple add_task calls — they run sequentially
• Sync functions run in a thread pool, async functions run in the event loop
• The client doesn't wait — the response arrives before the background tasks even start
• Common patterns: email notification, audit logging, data cleanup
• With Depends: you can encapsulate background task logic in reusable dependencies
• Conditional tasks: only call add_task if a condition is met (e.g. status → completed)
• Logging is essential — background tasks run silently; without logging you have no idea whether they failed

Additional resources

1. FastAPI - Background Tasks — The official tutorial with complete examples
2. Starlette - Background — The underlying implementation of BackgroundTasks
3. Python logging - Basic Tutorial — How to configure and use logging in Python
4. FastAPI - Dependencies — How to combine BackgroundTasks with Depends
5. Real Python - Logging — A practical guide to logging in Python

Next capsule: Patterns and error handling — You'll learn to chain tasks, handle errors in the background without affecting the client, and track the state of your operations.