A modern, async-first, type-safe task queue for Python 3.12+. Inspired by Laravel's elegant queue system. Native FastAPI integration. Switch between multiple queue backends (Redis, PostgreSQL, MySQL, RabbitMQ, AWS SQS) with one config line. Automatic ORM serialization (SQLAlchemy, Django, Tortoise) using msgpack reduces payloads by 90%+. Features ACID guarantees, dead-letter queues, crash recovery, and real-time event streaming.
📊 Looking for a monitoring dashboard? Check out asynctasq-monitor – a beautiful real-time UI to monitor your tasks, workers, and queues.
- AsyncTasQ
Unlike Celery and RQ which are built on synchronous foundations, AsyncTasQ is designed from the ground up for asyncio:
- Native async/await everywhere – All core operations use asyncio, no threading or blocking on critical paths
- Multiple execution modes – Choose between async I/O (event loop), sync I/O (thread pool), or CPU-bound (process pool) for each task
- High-performance concurrency – Process hundreds of tasks concurrently with minimal overhead using asyncio's efficient task scheduling
- Smart connection pooling – All drivers use connection pools optimized for async operations
- Non-blocking by design – Worker polling, task execution, and all I/O operations are truly non-blocking
- msgpack binary encoding – 2-3x faster than JSON with smaller payloads
- Automatic ORM model handling – Pass SQLAlchemy, Django ORM, or Tortoise ORM models directly as task arguments. AsyncTasQ automatically:
- Serializes them as lightweight references (primary key only)
- Reduces payload size by 90%+
- Re-fetches with fresh data when the task executes
- Supports all 3 major Python ORMs out of the box
- Smart type handling – Native support for
datetime,Decimal,UUID,set, and custom types without manual conversion
- ACID guarantees – PostgreSQL and MySQL drivers provide transactional dequeue with exactly-once processing semantics
- Built-in dead-letter queues – PostgreSQL/MySQL drivers automatically move permanently failed tasks to DLQ for inspection
- Crash recovery – Visibility timeouts ensure no task is lost even if workers crash mid-execution
- Graceful shutdown – SIGTERM/SIGINT handlers drain in-flight tasks before stopping (configurable timeout)
- Flexible retry strategies – Per-task retry configuration with custom
should_retry()hooks for intelligent retry logic - Task timeout protection – Prevent runaway tasks with configurable per-task timeouts
- Real-time observability – Redis Pub/Sub event streaming broadcasts task lifecycle events for monitoring dashboards
- Elegant, Laravel-inspired API – Clean, intuitive syntax that feels natural
- Full type safety – Complete type hints, mypy/pyright compatible, Generic
Task[T]for return type checking - Zero configuration – Works with environment variables out of the box, sensible defaults everywhere
- Two task styles – Choose function-based
@taskdecorators or class-based tasks with lifecycle hooks - Fluent method chaining – Configure tasks expressively:
.delay(60).on_queue("high").retry_after(120).dispatch() - First-class FastAPI integration – Lifespan management, automatic connection pooling, native async support
- 5 production-ready drivers – Redis, PostgreSQL, MySQL, RabbitMQ, AWS SQS – all with the same API
- Switch with one line – Change
driver="redis"todriver="postgres"– no code changes needed - Per-task driver override – Use Redis for high-throughput tasks, PostgreSQL for ACID-critical tasks in the same application
- Same API, different guarantees – Choose the driver that matches your SLA requirements without rewriting code
-
✅ Async-first design with asyncio throughout the stack
-
✅ Multiple queue drivers: Redis, PostgreSQL, MySQL, RabbitMQ, AWS SQS
-
✅ High-performance msgpack serialization with binary support
-
✅ Automatic ORM model handling for SQLAlchemy, Django, Tortoise
-
✅ Type-safe with full type hints and Generic support
-
✅ Three execution modes: Async (I/O), Thread pool (moderate CPU), Process pool (heavy CPU)
-
✅ Configurable retries with custom retry logic hooks
-
✅ Task timeouts to prevent runaway tasks
-
✅ Delayed task execution with precision timing
-
✅ Queue priority with multiple queues per worker
-
✅ Graceful shutdown with signal handlers
-
✅ ACID guarantees (PostgreSQL/MySQL drivers)
-
✅ Dead-letter queues for failed task inspection
-
✅ Visibility timeouts for crash recovery
-
✅ Connection pooling for optimal resource usage
-
✅ Transactional dequeue with
SELECT FOR UPDATE SKIP LOCKED -
✅ Task metadata tracking (attempts, timestamps, task IDs)
-
✅ Concurrent processing with configurable worker concurrency
-
✅ Real-time event streaming via Redis Pub/Sub
-
✅ FastAPI – Automatic lifecycle management, dependency injection
-
✅ SQLAlchemy – Async and sync model serialization
-
✅ Django ORM – Native async support (Django 3.1+)
-
✅ Tortoise ORM – Full async ORM integration
-
✅ asynctasq-monitor – Real-time monitoring dashboard (optional)
-
✅ Comprehensive CLI – Worker management and database migrations
-
✅ Function-based tasks with
@taskdecorator -
✅ Class-based tasks with 4 execution modes:
AsyncTask– Async I/O-bound (API calls, async DB queries)SyncTask– Sync I/O-bound via thread pool (requests, sync DB drivers)AsyncProcessTask– Async CPU-intensive via process poolSyncProcessTask– Sync CPU-intensive via process pool (bypasses GIL)
-
✅ Lifecycle hooks –
execute(),failed(),should_retry()for complete control -
✅ Method chaining for fluent task configuration
-
✅ Environment variable configuration for 12-factor apps
Get started in 60 seconds:
# Install AsyncTasQ (Python 3.12+ required)
uv add asynctasq[redis]import asyncio
from asynctasq.config import set_global_config
from asynctasq.tasks import task
# 1. Configure (or use environment variables)
set_global_config(driver="redis", redis_url="redis://localhost:6379")
# 2. Define a task
@task
async def send_email(to: str, subject: str, body: str):
print(f"Sending email to {to}: {subject}")
await asyncio.sleep(1) # Simulate email sending
return f"Email sent to {to}"
# 3. Dispatch the task
async def main():
for i in range(10):
task_id = await send_email.dispatch(
to=f"user{i}@example.com", subject=f"Welcome {i}!", body="Welcome to our platform!"
)
print(f"Task dispatched: {task_id}")
if __name__ == "__main__":
asyncio.run(main())# Run the worker (in a separate terminal)
python -m asynctasq workerThat's it! Your first AsyncTasQ is ready. Now let's explore the powerful features.
- One-line setup:
just init— install deps and pre-commit hooks - Start services:
just services-up— Redis, PostgreSQL, MySQL, RabbitMQ, LocalStack (SQS) for local integration tests - Run tests:
just test(orpytest) — usejust test-unit/just test-integrationto scope - Run with coverage:
just test-covorpytest --cov=src/asynctasq --cov-report=html - Run the worker locally:
python -m asynctasq worker - Pre-commit hooks:
./setup-pre-commit.shorjust setup-hooks - Format / lint / typecheck:
just format,just lint,just typecheck
- CI runs on PRs and pushes to
mainand includes lint, type checks and tests across Python 3.12–3.14. - Pre-commit hooks enforce formatting and static checks locally before commits (see
./setup-pre-commit.sh). - Branch protection: enable required status checks (CI success, lint, unit/integration jobs) for
main. - Coverage badge: the repository updates
.github/coverage.svgautomatically via.github/workflows/coverage-badge.yml. - Run full CI locally:
just ci(runs format/lint/typecheck/tests like the workflow).
| Feature | AsyncTasQ | Celery |
|---|---|---|
| Async Support | ✅ Async-first, built with asyncio | ❌ No asyncio support (promised for years, not delivered) |
| Type Safety | ✅ Full type hints, Generic[T] | |
| Multi-Driver | ✅ 5 drivers (Redis/PostgreSQL/MySQL/RabbitMQ/SQS) | |
| ORM Integration | ✅ Auto-serialization (SQLAlchemy/Django/Tortoise) | ❌ Manual serialization required |
| Serialization | ✅ msgpack (fast, binary, efficient) | |
| FastAPI Integration | ✅ First-class, lifespan management | |
| Dead-Letter Queue | ✅ Built-in (PostgreSQL/MySQL) | |
| ACID Guarantees | ✅ PostgreSQL/MySQL drivers | ❌ Not available |
| Global Rate Limiting | ❌ Not available (per-worker only) | |
| Setup Complexity | ✅ Zero-config with env vars | |
| Prefetch Multiplier | ✅ Sensible default (1) | |
| Learning Curve | ✅ Simple, intuitive API | |
| Maturity | ✅ 13+ years, battle-tested |
When to use AsyncTasQ:
- Modern async Python applications (FastAPI, aiohttp, async web frameworks)
- Need true asyncio support for I/O-bound tasks (API calls, database queries)
- Type-safe codebase with full IDE support
- Multiple driver flexibility (dev → production migration)
- Automatic ORM model handling (SQLAlchemy, Django, Tortoise)
- Enterprise ACID requirements (financial transactions, critical workflows)
- Simple, clean API without steep learning curve
When to use Celery:
- Mature ecosystem with many plugins and extensions
- Complex workflows (chains, chords, groups with callbacks)
- Large existing Celery codebase that's not worth migrating
- Synchronous applications where asyncio isn't needed
- Need for battle-tested, widely-adopted solution
| Feature | AsyncTasQ | Dramatiq |
|---|---|---|
| Async Support | ✅ Async-first, native asyncio | |
| Type Safety | ✅ Full type hints, Generic[T] | ✅ Type hints (py.typed) |
| Multi-Driver | ✅ 5 drivers (Redis/PostgreSQL/MySQL/RabbitMQ/SQS) | |
| ORM Integration | ✅ Auto-serialization (SQLAlchemy/Django/Tortoise) | ❌ Manual serialization required |
| Dead-Letter Queue | ✅ Built-in (PostgreSQL/MySQL) | ✅ Built-in (all brokers) |
| FastAPI Integration | ✅ First-class, lifespan management | |
| Database Drivers | ✅ PostgreSQL/MySQL with ACID | ❌ Not available |
| Simplicity | ✅ Clean, intuitive API | ✅ Simple, well-designed |
When to use AsyncTasQ:
- Async-first applications (FastAPI, aiohttp, modern Python stack)
- True asyncio support for I/O-bound tasks
- Database-backed queues with ACID guarantees
- Automatic ORM model serialization
- Type-safe codebase with IDE support
When to use Dramatiq:
- Synchronous applications
- Mature, battle-tested solution needed
- Complex middleware requirements
- Don't need async support
| Feature | AsyncTasQ | ARQ |
|---|---|---|
| Async Support | ✅ Async-first, native asyncio | ✅ Async-first, native asyncio |
| Multi-Driver | ✅ 5 drivers (Redis/PostgreSQL/MySQL/RabbitMQ/SQS) | ❌ Redis only |
| Type Safety | ✅ Full type hints, Generic[T] | ✅ Type hints |
| ORM Integration | ✅ Auto-serialization (SQLAlchemy/Django/Tortoise) | ❌ Manual serialization |
| Serialization | ✅ msgpack (binary, efficient) | |
| Dead-Letter Queue | ✅ Built-in (PostgreSQL/MySQL) | ❌ Not available |
| ACID Guarantees | ✅ PostgreSQL/MySQL drivers | ❌ Not available |
| FastAPI Integration | ✅ First-class, lifespan management | |
| Task Execution Model | ✅ At-least-once with idempotency support | |
| Simplicity | ✅ Clean, Laravel-inspired API | ✅ Lightweight, minimal |
| Custom Serializers | ✅ Configurable serializers | ✅ Configurable serializers |
When to use AsyncTasQ:
- Need multiple driver options (not locked into Redis)
- Database-backed queues with ACID guarantees
- Automatic ORM model handling
- Dead-letter queue support for failed task inspection
- FastAPI applications with first-class integration
- Enterprise reliability requirements
When to use ARQ:
- Simple Redis-only infrastructure
- Lightweight solution with minimal dependencies
- Cron job scheduling is a primary requirement
- Mature async task queue needed
- Custom serializers (e.g., msgpack) are acceptable to configure manually
| Feature | AsyncTasQ | RQ |
|---|---|---|
| Async Support | ✅ Async-first, native asyncio | ❌ Sync only (no asyncio support) |
| Multi-Driver | ✅ 5 drivers (Redis/PostgreSQL/MySQL/RabbitMQ/SQS) | ❌ Redis only |
| Type Safety | ✅ Full type hints, Generic[T] | ✅ Type hints added |
| Retries | ✅ Configurable with custom should_retry() |
✅ Configurable retries |
| Dead-Letter Queue | ✅ Built-in (PostgreSQL/MySQL) | ❌ Not available |
| Database Drivers | ✅ PostgreSQL/MySQL with ACID | ❌ Not available |
| Simplicity | ✅ Intuitive, clean API | ✅ Very simple |
When to use AsyncTasQ:
- Async applications (FastAPI, aiohttp)
- True asyncio support for efficient I/O
- Multiple driver options
- Enterprise features (DLQ, ACID)
- ORM integration
When to use RQ:
- Simple, synchronous use cases
- Synchronous applications
- Redis-only infrastructure
- Need mature, battle-tested solution
| Feature | AsyncTasQ | Huey |
|---|---|---|
| Async Support | ✅ Async-first, native asyncio | |
| Multi-Driver | ✅ 5 drivers (Redis/PostgreSQL/MySQL/RabbitMQ/SQS) | |
| Type Safety | ✅ Full type hints, Generic[T] | |
| ORM Integration | ✅ Auto-serialization (SQLAlchemy/Django/Tortoise) | ❌ Manual serialization |
| Enterprise Features | ✅ ACID, DLQ, visibility timeout | |
| Simplicity | ✅ Clean, modern API | ✅ Simple, lightweight |
| Cron Jobs | ✅ Built-in periodic tasks |
When to use AsyncTasQ:
- Async-first applications requiring true asyncio
- Enterprise requirements (ACID, DLQ)
- Type-safe codebase with IDE support
- Automatic ORM integration
- Need for multiple driver options
When to use Huey:
- Lightweight use cases
- Simple periodic/cron tasks
- SQLite-backed queues for embedded apps
- Mature, stable solution needed
AsyncTasQ stands out with:
- True async-first architecture – Built with asyncio from the ground up (unlike Celery, RQ, Huey)
- Multiple execution modes – Choose async I/O, sync I/O (thread pool), or CPU-bound (process pool) per task
- Intelligent ORM handling – Automatic model serialization for SQLAlchemy, Django ORM, and Tortoise ORM (90%+ smaller payloads)
- msgpack serialization – Binary format that's 2-3x faster than JSON with smaller payloads
- Multi-driver flexibility – 5 production-ready drivers (Redis, PostgreSQL, MySQL, RabbitMQ, AWS SQS) with identical API
- Type safety everywhere – Full type hints with Generic[T] support, mypy/pyright compatible
- Enterprise ACID guarantees – PostgreSQL/MySQL drivers with transactional dequeue for exactly-once processing
- Built-in dead-letter queues – PostgreSQL/MySQL drivers automatically handle permanently failed tasks
- First-class FastAPI integration – Lifespan management, automatic connection pooling, native async support
- Real-time event streaming – Redis Pub/Sub broadcasts task lifecycle events for monitoring
- Optional monitoring UI – Beautiful real-time dashboard via asynctasq-monitor
- Elegant, Laravel-inspired API – Method chaining (
.delay(60).on_queue("high").dispatch()) and intuitive task definitions - Zero configuration – Works with environment variables out of the box, sensible defaults everywhere
A beautiful real-time monitoring dashboard for AsyncTasQ:
- 📈 Live Dashboard – Real-time task metrics, queue depths, and worker status
- 📊 Task Analytics – Execution times, success/failure rates, retry patterns
- 🔍 Task Explorer – Browse, search, and inspect task details
- 👷 Worker Management – Monitor worker health and performance
- 🚨 Alerts – Get notified about failures and queue backlogs
# Install the monitoring package
uv add asynctasq-monitor
# Start the monitoring server
asynctasq-monitor webFor detailed documentation, see the following guides:
- Installation – Installation instructions for uv and pip
- Queue Drivers – Redis, PostgreSQL, MySQL, RabbitMQ, AWS SQS
- ORM Integrations – SQLAlchemy, Django, Tortoise ORM
- Framework Integrations – FastAPI integration
- Task Definitions – Function-based and class-based tasks
- Running Workers – CLI and programmatic workers
- Configuration – Environment variables, programmatic, CLI
- CLI Reference – Complete command reference
- Best Practices – Task design, queue organization, production deployment
For complete examples, see the following guides:
- Function-Based Tasks Examples – Complete examples guide
- Class-Based Tasks Examples – Complete examples guide
Contributions are welcome! Please see CONTRIBUTING.md for guidelines.
MIT License – see LICENSE file for details.
- Repository: github.com/adamrefaey/asynctasq
- Issues: github.com/adamrefaey/asynctasq/issues
- Discussions: github.com/adamrefaey/asynctasq/discussions
- SQLite driver support
- Oracle driver support
- Task batching support
- Task chaining and workflows (chains, chords, groups)
- Rate limiting
- Task priority within queues
- Scheduled/cron tasks
Built with ❤️ by Adam Refaey.