Background tasks in FastAPI | Python Backend

Background tasks in FastAPI: what developers should know

You've just processed a user's payment, confirmed their order, and now you need to send a confirmation email. But do you really want to make the customer wait while you wait for your email service to respond? Background tasks in FastAPI let you return a response immediately while work continues in the background. Understanding when and how to use them—and, critically, when not to—separates production-ready systems from ones that lose data and frustrate users.

What are background tasks?

FastAPI's BackgroundTasks is a built-in utility that lets you queue work to run after the HTTP response is sent back to the client. The endpoint returns quickly, giving the user immediate feedback, while the server handles ancillary work in the background.

Here's a simple example:

python

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def send_email(email: str, subject: str):
    print(f"Sending {subject} to {email}")
    # Simulate slow email service
    import time
    time.sleep(5)
    print(f"Email sent to {email}")

@app.post("/orders")
async def create_order(order_data: dict, background_tasks: BackgroundTasks):
    # Process the order synchronously
    order_id = save_order_to_db(order_data)
    
    # Queue the email to send after response
    background_tasks.add_task(
        send_email, 
        email=order_data["customer_email"],
        subject=f"Order {order_id} confirmed"
    )
    
    return {"order_id": order_id, "status": "confirmed"}

The client receives {"order_id": 123, "status": "confirmed"} in milliseconds, while the email task runs on the server. No waiting.

BackgroundTasks for small fire-and-forget jobs

FastAPI's BackgroundTasks is perfect for genuinely optional work that doesn't need guarantees. Think of it as the scratchpad of background work—lightweight, convenient, and always available.

Good candidates for BackgroundTasks:

Sending welcome emails or reminders
Logging analytics events to a data warehouse
Updating user profiles with non-critical metadata
Pruning old cache entries or cleaning temporary files
Notifying third-party services of non-critical events

All these jobs share a trait: if they fail silently, the system still works. The user still got their order, the analytics are just slightly incomplete, the cache is a bit messier.

Here's a more realistic example:

python

from fastapi import BackgroundTasks, FastAPI
from datetime import datetime
import httpx

app = FastAPI()

async def log_analytics(user_id: int, action: str):
    """Non-critical analytics logging"""
    async with httpx.AsyncClient() as client:
        await client.post(
            "https://analytics.internal/events",
            json={
                "user_id": user_id,
                "action": action,
                "timestamp": datetime.utcnow().isoformat()
            },
            timeout=5.0  # Don't hang the background task
        )

async def refresh_user_cache(user_id: int):
    """Update user's cached profile after they change settings"""
    user = get_user_from_db(user_id)
    cache.set(f"user:{user_id}", user, ttl=3600)

@app.post("/users/{user_id}/settings")
async def update_user_settings(
    user_id: int, 
    settings: dict,
    background_tasks: BackgroundTasks
):
    update_db(user_id, settings)
    
    # Queue non-blocking work
    background_tasks.add_task(log_analytics, user_id, "settings_updated")
    background_tasks.add_task(refresh_user_cache, user_id)
    
    return {"status": "settings updated"}

Notice the pattern: we save the critical data first, then queue the rest. The response goes out immediately, and these tasks run when the event loop gets to them.

Use Celery or a queue for anything important

The moment your task must not be lost, must be retried on failure, or must survive a server restart, you've outgrown BackgroundTasks. Move to a proper task queue like Celery, RQ, or Temporal.

Critical work includes:

Payment processing or refunds
Sending transactional emails (password resets, invoices)
Syncing data to external systems (payment processors, CRMs, reporting platforms)
Generating reports or exports that customers depend on
Scheduling recurring actions (weekly digests, subscription renewals)

With Celery and Redis, your task gets stored in a broker before execution. If the worker crashes, the task stays in Redis and gets retried. Your FastAPI app just enqueues it and moves on.

python

from celery import Celery
from fastapi import FastAPI

celery_app = Celery("tasks", broker="redis://localhost:6379")
app = FastAPI()

@celery_app.task(bind=True, max_retries=3)
def process_payment(self, payment_id: int):
    try:
        payment = get_payment_from_db(payment_id)
        stripe.Charge.create(
            amount=payment.amount,
            currency="usd",
            source=payment.source_token
        )
        mark_payment_complete(payment_id)
    except Exception as exc:
        # Retry with exponential backoff (5s, 25s, 125s)
        raise self.retry(exc=exc, countdown=5 ** self.request.retries)

@app.post("/payments")
async def create_payment(payment_data: dict):
    payment_id = save_payment_to_db(payment_data)
    
    # Enqueue to Celery, not to BackgroundTasks
    process_payment.delay(payment_id)
    
    return {"payment_id": payment_id, "status": "processing"}

Now if your worker dies mid-payment, Redis holds the task. When the worker restarts, it processes the payment. With BackgroundTasks, that payment would simply vanish.

Tasks die with the process — no retries

This is the critical limitation of BackgroundTasks: once you queue a task, it lives only in the current Python process's memory. If the process crashes, restarts, or shuts down gracefully before the task completes, the task is gone.

Imagine this scenario:

python

@app.post("/newsletter")
async def subscribe(email: str, background_tasks: BackgroundTasks):
    save_subscriber(email)
    
    # Queue a welcome email
    background_tasks.add_task(send_welcome_email, email)
    
    return {"status": "subscribed"}

# Meanwhile, someone deploys a new version. 
# The process shuts down while send_welcome_email is still running.
# No retry logic, no persistence. The email never sent.

For non-critical work (like analytics), this is acceptable. For anything transactional, it's a bug waiting to happen.

If you do use BackgroundTasks, add defensive timeouts so a slow task doesn't block graceful shutdown:

python

import asyncio

async def send_notification(user_id: int, message: str):
    try:
        # Wrap in timeout so it doesn't block forever
        await asyncio.wait_for(
            notify_user_service(user_id, message),
            timeout=10.0
        )
    except asyncio.TimeoutError:
        logger.warning(f"Notification timeout for user {user_id}")

Never put payment logic in background tasks

This deserves its own section because we've all seen it go wrong. Payment processing, subscription renewals, and any financial transaction must never rely on BackgroundTasks.

Here's why:

Charges disappear silently if the process crashes
No built-in retry mechanism means failed payments don't get reattempted
No visibility into what succeeded and what didn't
No dead-letter queue for failed transactions to investigate
Auditors will have questions your logs can't answer

The wrong way (never do this):

python

@app.post("/checkout")
async def checkout(order: Order, background_tasks: BackgroundTasks):
    order_id = create_order(order)
    
    # WRONG: Payment disappears if process crashes
    background_tasks.add_task(charge_credit_card, order.card_token, order.total)
    
    return {"order_id": order_id}

The right way (use a task queue):

python

@app.post("/checkout")
async def checkout(order: Order):
    order_id = create_order(order, status="pending_payment")
    
    # Enqueue to Celery with retries and persistence
    process_payment_task.delay(order_id, order.card_token, order.total)
    
    return {"order_id": order_id, "status": "payment processing"}

@celery_app.task(bind=True, max_retries=5)
def process_payment_task(self, order_id: int, token: str, amount: float):
    try:
        charge_id = charge_credit_card(token, amount)
        mark_order_paid(order_id, charge_id)
    except PaymentFailedError as exc:
        if self.request.retries < self.max_retries:
            # Retry with exponential backoff
            raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))
        else:
            # Log failure, notify team, mark order as failed
            mark_order_failed(order_id)
            notify_payment_team(order_id, str(exc))

Common mistakes

Mistake 1: Treating BackgroundTasks as a task queue. It's not. It has no persistence, no retries, no monitoring. Use it for convenience, not reliability.

Mistake 2: Blocking background tasks with synchronous operations. If you do use BackgroundTasks, make them async so they don't starve the event loop. Blocking calls like requests.get() will freeze your FastAPI app while the background task runs.

python

# WRONG: Synchronous HTTP call blocks the event loop
def send_webhook(url: str, data: dict):
    requests.post(url, json=data)  # Blocks!

background_tasks.add_task(send_webhook, "https://...", data)

# RIGHT: Use an async HTTP client
async def send_webhook(url: str, data: dict):
    async with httpx.AsyncClient() as client:
        await client.post(url, json=data)

background_tasks.add_task(send_webhook, "https://...", data)

Mistake 3: Not setting timeouts on background tasks. A hanging task can exhaust your worker pool and cascade into production outages. Always add timeouts:

python

async def risky_operation():
    try:
        return await asyncio.wait_for(external_api_call(), timeout=5.0)
    except asyncio.TimeoutError:
        logger.error("Operation timed out")
        return None

Mistake 4: Assuming background tasks run in a specific order. FastAPI queues them, but they run whenever the event loop gets to them. Don't chain dependencies between tasks.

Mistake 5: Not logging what happened. If a background task fails silently, you won't know. Always log entry and exit:

python

import logging

logger = logging.getLogger(__name__)

async def important_cleanup(resource_id: int):
    logger.info(f"Starting cleanup for resource {resource_id}")
    try:
        await perform_cleanup(resource_id)
        logger.info(f"Cleanup complete for resource {resource_id}")
    except Exception as e:
        logger.error(f"Cleanup failed for resource {resource_id}: {e}")

Key takeaways

Use BackgroundTasks only for small, optional work that can safely fail silently—analytics, welcome emails, cache updates.
For anything critical, use a proper task queue (Celery, RQ, Temporal) that persists tasks and retries on failure.
Never put payment processing, subscription renewals, or transactional emails in BackgroundTasks.
BackgroundTasks die with the process. There's no recovery, no retries, no queue. Keep them simple and disposable.
Always use async code in background tasks and add timeouts so they don't block your event loop or graceful shutdown.
Log everything. If it's not in the logs, it never happened.

Background tasks in FastAPI: what developers should know

Background tasks in FastAPI: what developers should know

What are background tasks?

BackgroundTasks for small fire-and-forget jobs

Use Celery or a queue for anything important

Tasks die with the process — no retries

Never put payment logic in background tasks

Common mistakes

Key takeaways

Comments

Leave a comment

Related articles

Why I still choose .NET for backends: a practical guide

Minimal APIs: a production-ready setup, explained simply

Program.cs, line by line: a hands-on guide for engineers

DI lifetimes: Singleton, Scoped, Transient — a deep dive