FastAPI: Background work & task queues

Some work should not make the client wait: sending an email, calling a webhook, generating a report. FastAPI’s BackgroundTasks run a function after the response is sent, in the same process. That is perfect for small, best-effort jobs, and dangerous for anything that must happen, because a crash, a deploy or an exception silently loses the work while the client already has its 200. Work that matters goes to a durable queue, often through a transactional outbox, processed idempotently with retries. This part builds all of it.

1. BackgroundTasks: after the response

the response goes out first; the task runs afterwards

import time

from fastapi import BackgroundTasks, FastAPI
from fastapi.testclient import TestClient

timeline = []
start = time.perf_counter()
def mark(event):
    timeline.append(f"{(time.perf_counter() - start) * 1000:6.0f} ms  {event}")

def send_receipt(email: str, order_id: int):
    time.sleep(0.3)                                     # a slow SMTP call
    mark(f"task done: receipt for order {order_id} sent to {email}")

app = FastAPI()

@app.middleware("http")
async def response_watch(request, call_next):
    response = await call_next(request)
    mark(f"response {response.status_code} handed to the server")
    return response

@app.post("/orders", status_code=201)
def create_order(background: BackgroundTasks):
    mark("endpoint: order saved")
    background.add_task(send_receipt, "[email protected]", 1042)    # def or async def both work
    return {"id": 1042}

TestClient(app).post("/orders")
print("\n".join(timeline))

Output

    92 ms  endpoint: order saved
    92 ms  response 201 handed to the server
   393 ms  task done: receipt for order 1042 sent to [email protected]

The client got its 201 without waiting for the 300 ms email. That is the whole point. Now the failure modes:

a failing background task: the client never finds out

import logging

from fastapi import BackgroundTasks, FastAPI
from fastapi.testclient import TestClient

logging.disable(logging.CRITICAL)                        # keep the demo output short
def charge_loyalty_points(user_id: int):
    raise ConnectionError("loyalty service timed out")

app = FastAPI()

@app.post("/checkout")
def checkout(background: BackgroundTasks):
    background.add_task(charge_loyalty_points, 7)
    return {"status": "paid"}

client = TestClient(app, raise_server_exceptions=False)
r = client.post("/checkout")
print("client saw:", r.status_code, r.json(), "<- the points were never added, and nothing will retry")

Output

client saw: 200 {'status': 'paid'} <- the points were never added, and nothing will retry

2. The transactional outbox

The classic bug: save the order in the database, then publish “order created” to a queue. If the process dies between the two steps, the order exists but nothing downstream ever hears about it. Publish first and die, and downstream hears about an order that does not exist. The outbox pattern fixes this with one transaction: write the business row and an outbox row together, then let a separate relay deliver outbox rows, retry failures, and mark them done.

outbox + relay + idempotent consumer, with a crash in the middle

import json
import random
import sqlite3
import uuid

db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INT);
CREATE TABLE outbox (event_id TEXT PRIMARY KEY, payload TEXT, status TEXT DEFAULT 'pending', attempts INT DEFAULT 0);
""")

def create_order(sku, qty):
    with db:                                              # ONE transaction: both rows or neither
        order_id = db.execute("INSERT INTO orders (sku, qty) VALUES (?, ?)", (sku, qty)).lastrowid
        event = {"type": "order.created", "order_id": order_id, "sku": sku, "qty": qty}
        db.execute("INSERT INTO outbox (event_id, payload) VALUES (?, ?)", (uuid.uuid4().hex, json.dumps(event)))
    return order_id

processed_event_ids = set()                               # the consumer's dedupe store (a DB table in real life)
warehouse_stock = {"TEA-01": 100}
def consumer(event_id, event):                            # e.g. the warehouse service
    if event_id in processed_event_ids:
        return "duplicate ignored"
    warehouse_stock[event["sku"]] -= event["qty"]
    processed_event_ids.add(event_id)
    return "applied"

random.seed(4)
def relay_once(crash_after_send=False):
    rows = db.execute("SELECT event_id, payload FROM outbox WHERE status='pending' ORDER BY rowid").fetchall()
    for event_id, payload in rows:
        db.execute("UPDATE outbox SET attempts = attempts + 1 WHERE event_id=?", (event_id,))
        if random.random() < 0.3:                        # the network sometimes fails
            print(f"  send {event_id[:6]} failed, will retry")
            continue
        result = consumer(event_id, json.loads(payload))
        print(f"  send {event_id[:6]} → consumer: {result}")
        if crash_after_send:
            print("  💥 relay crashed before marking it sent")
            return
        db.execute("UPDATE outbox SET status='sent' WHERE event_id=?", (event_id,))
    db.commit()

for qty in (2, 5, 1):
    create_order("TEA-01", qty)
print("run 1 (crashes after the first successful send):")
relay_once(crash_after_send=True)
for n in range(2, 5):
    print(f"run {n}:")
    relay_once()
left = db.execute("SELECT COUNT(*) FROM outbox WHERE status='pending'").fetchone()[0]
print(f"pending events: {left} | warehouse stock: {warehouse_stock['TEA-01']} (100 - 2 - 5 - 1 = 92)")

Output

run 1 (crashes after the first successful send):
  send 34f675 failed, will retry
  send 0b77dd failed, will retry
  send 4d82df → consumer: applied
  💥 relay crashed before marking it sent
run 2:
  send 34f675 failed, will retry
  send 0b77dd failed, will retry
  send 4d82df → consumer: duplicate ignored
run 3:
  send 34f675 → consumer: applied
  send 0b77dd → consumer: applied
run 4:
pending events: 0 | warehouse stock: 92 (100 - 2 - 5 - 1 = 92)

The relay crashed after delivering an event but before recording it, so it delivered the same event again on the next run. The consumer recognised the event id and ignored the duplicate. That is the combination to aim for: at-least-once delivery, idempotent processing. In production the relay is a small worker (or a change-data-capture stream), and the destination is a queue or broker such as Redis, RabbitMQ, SQS, Kafka or Pub/Sub.

3. Choosing a task queue

4. Retrying a flaky call with backoff and jitter

tenacity: retry only what can succeed later

import random

from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential_jitter

class TransientError(Exception): ...
class PermanentError(Exception): ...

random.seed(1)
attempts = []

@retry(retry=retry_if_exception_type(TransientError),          # never retry a permanent failure
       wait=wait_exponential_jitter(initial=0.01, max=0.2),    # 10 ms, 20 ms, 40 ms … plus jitter
       stop=stop_after_attempt(5), reraise=True)
def deliver_webhook(url, payload):
    attempts.append(url)
    if random.random() < 0.6:
        raise TransientError("503 from partner")
    return 200

print("delivered:", deliver_webhook("https://partner.example/hooks", {"order": 1042}),
      "after", len(attempts), "attempts")

attempts.clear()
@retry(retry=retry_if_exception_type(TransientError), stop=stop_after_attempt(5), reraise=True)
def bad_request():
    attempts.append(1)
    raise PermanentError("400: invalid signature")
try:
    bad_request()
except PermanentError as e:
    print("permanent failure, attempts:", len(attempts), "-", e)

Output

delivered: 200 after 2 attempts
permanent failure, attempts: 1 - 400: invalid signature

Recap