🇬🇧 fastapi-arq – FastAPI + ARQ without boilerplate

**🇬🇧 English** · [🇷🇺 Русский](README.ru.md) · [🇨🇳 简体中文](README.zh-CN.md) · [🇹🇼 繁體中文](README.zh-TW.md) · [🇪🇸 Español](README.es.md) · [🇫🇷 Français](README.fr.md) · [🇩🇪 Deutsch](README.de.md) · [🇯🇵 日本語](README.ja.md) · [🇰🇷 한국어](README.ko.md) · [🇸🇦 العربية](README.ar.md) · [🇵🇹 Português](README.pt.md) · [🇮🇹 Italiano](README.it.md) · [🇳🇱 Nederlands](README.nl.md) · [🇵🇱 Polski](README.pl.md) · [🇹🇷 Türkçe](README.tr.md) · [🇻🇳 Tiếng Việt](README.vi.md) · [🇹🇭 ไทย](README.th.md) · [🇮🇩 Bahasa Indonesia](README.id.md) · [🇮🇳 हिन्दी](README.hi.md) · [🇺🇦 Українська](README.uk.md)

📖 About

fastapi-arq is a lightweight decorator-style wrapper that integrates arq (an in-process async Redis queue) into FastAPI applications. It eliminates repetitive boilerplate: starting/stopping the worker via FastAPI lifespan, injecting DI dependencies (Depends(...)) into task functions, typed job enqueuing, and a built-in health-check endpoint.

🔍 The Problem

Every FastAPI project that uses arq ends up writing the same glue code: manually starting/stopping arq.Worker in the lifespan, creating Redis connection pools, injecting FastAPI dependencies (DB session, S3 client) into worker tasks, writing health-check endpoints, and handling unique job scheduling, retries, and timeouts. This code is copied from project to project, leading to fragmentation and maintenance burden.

✅ The Solution

fastapi-arq is the glue between FastAPI and arq — not a replacement. One-liner setup: FastArq(app, 'redis://...'). Declarative tasks: @arq.task(max_tries=3). DI injection: Depends(get_db) in worker context. Typed enqueue: arq_redis.enqueue(my_task, user_id=42). Built-in health: /arq/health, /live, /ready.

🧩 Components

📦 Requirements

• Python 3.11+
• FastAPI ≥ 0.111.0
• arq ≥ 0.28.0
• Redis server (any version ≥ 6)

🛠️ Installation

pip install fastapi-arq
# or with dev extras
pip install "fastapi-arq[dev]"

🚀 Quick Start

from fastapi import FastAPI, Depends
from fastapi_arq import FastArq

app = FastAPI()

arq = FastArq(app, redis_settings="redis://localhost:6379")

@arq.task(max_tries=3, timeout=120)
async def send_welcome_email(ctx: dict, user_id: int):
    db = ctx["db"]
    user = await db.get(User, user_id)
    ...

@app.post("/users")
async def create_user(arq_redis=Depends(arq.get_redis)):
    ...
    await arq_redis.enqueue(send_welcome_email, user_id=user.id)
    return {"ok": True}

app.include_router(arq.health_router)

🆕 v0.2.0 Features

Cron jobs

@arq.cron(hour=6, minute=0)
async def daily_report(ctx: dict) -> None:
    ...

@arq.cron(hour={9, 18}, minute=0, run_at_startup=True)
async def sync_data(ctx: dict) -> None:
    ...

External worker (production)

# app.py — enqueue-only process
arq = FastArq(app, "redis://...", run_worker=False)

# worker.py — standalone worker process
import asyncio
from myapp import arq
asyncio.run(arq.worker_runner.run())

📋 FAQ

Q: Does fastapi-arq replace arq?

No. fastapi-arq is a thin layer on top of arq. You still need arq and Redis. Think of it as 'FastAPI sugar for arq' — it handles the wiring so you don't have to.

Q: Can I use my own existing lifespan?

Yes. FastArq preserves and wraps any existing lifespan context manager. Your startup/shutdown logic runs alongside the worker lifecycle.

Q: How do I pass a DB session to tasks?

Pass context_dependencies=[Depends(get_db)] to FastArq. The builder resolves the dependency at worker startup and puts it in the ctx dict. Access as ctx['get_db'] in your task. Supports sync, async, and async-generator dependencies.

Q: Does fastapi-arq run workers in separate processes?

Yes. Pass run_worker=False to FastArq(app, ..., run_worker=False) and run arq.worker_runner.run() in a separate process. See External Worker.

💡 Common Patterns & Tips

• Unique jobs: Use enqueue_unique(my_task, arg=val) to prevent duplicate scheduling. Uses SETNX with MD5-hashed arguments and configurable TTL.
• Graceful shutdown: The worker._shutdown event is triggered on cancel; ongoing jobs are awaited via asyncio.gather with a 10-second timeout.
• Context naming: The dependency key comes from name of the callable: Depends(get_db) → ctx['get_db'].
• Cron jobs: Use @arq.cron(hour=6, minute=0) for periodic tasks. Also registered as regular tasks for manual enqueue.
• External worker: Production deployment with run_worker=False + arq.worker_runner.run() in a separate process.
• Testing: Use fakeredis for offline testing. The library includes pytest fixtures and async test helpers.

📚 Guides

Detailed documentation for every component and feature:

• API Reference — classes, methods, parameters
• Lifespan — lifecycle management, custom lifespan, shutdown order
• Context Dependencies — injecting DB sessions, clients into tasks
• Unique Jobs — preventing duplicate job scheduling
• Graceful Shutdown — cancellation, timeouts, cleanup
• Testing — offline testing with fakeredis, pytest fixtures
• Cron Jobs — periodic task scheduling
• External Worker — running the worker in a separate process
• FAQ & Troubleshooting — known issues, workarounds

📄 License

MIT — use freely in personal and commercial projects.