Run background jobs from FastAPI with ARQ

Approach

Use BackgroundTasks only for small, best-effort work. For durable jobs, enqueue to a Redis-backed queue like ARQ (asyncio-native) after committing your transaction, return 202 Accepted with a job/status ID, and run workers as a separate process with retries.

Solution

# app/jobs.py
from arq import Retry
from arq.connections import RedisSettings


async def send_receipt(ctx, order_id: int) -> None:
    async with SessionLocal() as db:
        order = await db.get(Order, order_id)
        if order is None or order.receipt_sent_at:
            return                                     # idempotent
        try:
            await mailer.send_receipt(order)
        except TemporaryMailError:
            raise Retry(defer=ctx["job_try"] * 30)    # backoff
        order.receipt_sent_at = utcnow()
        await db.commit()


class WorkerSettings:
    functions = [send_receipt]
    redis_settings = RedisSettings.from_dsn("redis://redis:6379/1")
    max_tries = 6


# app/main.py
from contextlib import asynccontextmanager
from arq import create_pool


@asynccontextmanager
async def lifespan(app):
    app.state.arq = await create_pool(WorkerSettings.redis_settings)
    yield
    await app.state.arq.close()


@app.post("/orders", status_code=201)
async def create_order(body: OrderIn, db: DB, request: Request):
    order = await orders.create(db, body)
    await db.commit()                                   # commit BEFORE enqueuing
    await request.app.state.arq.enqueue_job("send_receipt", order.id)
    return {"id": order.id}

# Run workers: arq app.jobs.WorkerSettings

Pitfalls

• Enqueuing before commit (the worker can't find the row).
• Non-idempotent jobs under at-least-once delivery.
• Using the request's DB session inside the job.

Variations

• Celery for a mature ecosystem (beat scheduler, canvas workflows).
• A transactional outbox table when enqueue must be atomic with the DB write.

Further reading: ARQ documentation