Metadata-Version: 2.4
Name: fastapi-arq
Version: 0.2.0
Summary: FastAPI + ARQ: async worker queue integration without boilerplate
Author-email: Artem Sitdikov <artem.sitd@gmail.com>
License-Expression: MIT
Project-URL: homepage, https://github.com/artem-sitd/fastapi-arq
Project-URL: repository, https://github.com/artem-sitd/fastapi-arq
Project-URL: documentation, https://github.com/artem-sitd/fastapi-arq/tree/main/docs/guides
Project-URL: changelog, https://github.com/artem-sitd/fastapi-arq/blob/main/CHANGELOG.md
Project-URL: issues, https://github.com/artem-sitd/fastapi-arq/issues
Keywords: fastapi,arq,redis,queue,worker,async
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Framework :: FastAPI
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: fastapi>=0.111.0
Requires-Dist: arq>=0.28.0
Requires-Dist: pydantic>=2.0
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.24; extra == "dev"
Requires-Dist: mypy>=1.10; extra == "dev"
Requires-Dist: ruff>=0.5; extra == "dev"
Requires-Dist: fakeredis>=2.24; extra == "dev"
Requires-Dist: httpx>=0.27; extra == "dev"
Provides-Extra: test
Requires-Dist: pytest>=8.0; extra == "test"
Requires-Dist: pytest-asyncio>=0.24; extra == "test"
Requires-Dist: fakeredis>=2.24; extra == "test"
Requires-Dist: httpx>=0.27; extra == "test"
Dynamic: license-file

<h1 align="center">🇬🇧 fastapi-arq – FastAPI + ARQ without boilerplate</h1>

<p align="center">**🇬🇧 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)</p>

---

![python](https://img.shields.io/badge/python-3.11_2B2B2B?logo=python) ![fastapi](https://img.shields.io/badge/fastapi-0.111_2B2B2B?logo=fastapi) ![arq](https://img.shields.io/badge/arq-0.28_2B2B2B?logo=redis) ![license](https://img.shields.io/badge/license-MIT-2B2B2B) ![tests](https://img.shields.io/badge/tests-114_passing-2B2B2B?logo=pytest)

---

## 📖 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

| Component | Purpose |
|---|---|
| `FastArq` | Main class. Registers workers, manages lifespan, exposes health router. |
| `@arq.task` | Decorator. Registers functions + settings (max_tries, timeout, keep_result, unique). |
| `@arq.cron` | Decorator. Registers periodic jobs (month, day, hour, minute — same params as arq). |
| `ArqRedis` | Extended arq.connections.ArqRedis with typed enqueue() and enqueue_unique(). |
| `ContextBuilder` | Resolves Depends(...) into a dict passed to every task as ctx. |
| `WorkerManager` | Starts/stops arq.Worker as an async task. Cancellation-safe. |
| `ArqWorker` / `WorkerSettings` | Run the worker in a separate process for production isolation. |
| `HealthRouter` | FastAPI APIRouter with /health (ping + latency), /live, /ready endpoints. |

## 📦 Requirements

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

## 🛠️ Installation

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

## 🚀 Quick Start

```python
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)
```

## 📋 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](docs/guides/external-worker.md).

## 💡 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](docs/guides/api-reference.md) — classes, methods, parameters
- [Lifespan](docs/guides/lifespan.md) — lifecycle management, custom lifespan, shutdown order
- [Context Dependencies](docs/guides/context-dependencies.md) — injecting DB sessions, clients into tasks
- [Unique Jobs](docs/guides/unique-jobs.md) — preventing duplicate job scheduling
- [Graceful Shutdown](docs/guides/graceful-shutdown.md) — cancellation, timeouts, cleanup
- [Testing](docs/guides/testing.md) — offline testing with fakeredis, pytest fixtures
- [Cron Jobs](docs/guides/cron.md) — periodic task scheduling
- [External Worker](docs/guides/external-worker.md) — running the worker in a separate process
- [FAQ & Troubleshooting](docs/guides/faq.md) — known issues, workarounds

## 📄 License
MIT — use freely in personal and commercial projects.
