TaskQueue Pro 🔄

Enterprise-grade distributed task queue system with scheduling, monitoring, and auto-scaling capabilities.

🌟 Features

Core Features

• ✅ Distributed Task Processing - Scale workers horizontally across multiple machines
• ✅ Priority Queues - High/medium/low priority task routing
• ✅ Task Scheduling - Cron-like scheduling with timezone support
• ✅ Retry Mechanisms - Exponential backoff, max retries, dead letter queues
• ✅ Result Backend - Store task results in Redis, PostgreSQL, or S3
• ✅ Task Chains & Groups - Compose complex workflows
• ✅ Rate Limiting - Control task execution rate per worker
• ✅ Auto-scaling - Scale workers based on queue depth

Backends Supported

Backend	Type	Use Case
Redis	Broker + Result	High performance, low latency
RabbitMQ	Broker	Reliable message delivery
PostgreSQL	Result	Persistent result storage
Amazon SQS	Broker	Cloud-native, serverless
Kafka	Broker	High throughput streaming

Worker Features

• ✅ Multi-processing - Process multiple tasks concurrently
• ✅ Auto-reload - Restart workers on code changes
• ✅ Health Checks - Monitor worker health and restart if needed
• ✅ Prefetch Multiplier - Control task prefetching
• ✅ Task Time Limits - Hard and soft timeouts
• ✅ Worker Pools - ThreadPool, ProcessPool, GreenletPool

Monitoring

• ✅ Web Dashboard - Real-time task and worker monitoring
• ✅ Prometheus Metrics - Export metrics for Grafana
• ✅ Task Tracing - Track task execution flow
• ✅ Alerts - Get notified on failures
• ✅ Log Aggregation - Centralized logging

🏗️ Architecture

┌─────────────────────────────────────────────────────────────────┐
│                      TASKQUEUE PRO                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐     │
│  │   Client     │    │   Client     │    │   Client     │     │
│  │   (API)      │    │   (CLI)      │    │   (SDK)      │     │
│  └──────┬───────┘    └──────┬───────┘    └──────┬───────┘     │
│         │                    │                    │              │
│         └────────────────────┼────────────────────┘              │
│                              ▼                                  │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                   BROKER (Redis/RabbitMQ)                │   │
│  │  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐       │   │
│  │  │ High Priority│ │Med Priority │ │Low Priority │       │   │
│  │  │   Queue      │ │   Queue     │ │   Queue     │       │   │
│  │  └─────────────┘ └─────────────┘ └─────────────┘       │   │
│  └──────────────────────────┬──────────────────────────────┘   │
│                             │                                   │
│         ┌───────────────────┼───────────────────┐              │
│         ▼                   ▼                   ▼              │
│  ┌──────────────┐   ┌──────────────┐   ┌──────────────┐      │
│  │   Worker 1   │   │   Worker 2   │   │   Worker N   │      │
│  │  (Process)   │   │  (Process)   │   │  (Process)   │      │
│  │              │   │              │   │              │      │
│  │ ┌──────────┐ │   │ ┌──────────┐ │   │ ┌──────────┐ │      │
│  │ │ Prefetch │ │   │ │ Prefetch │ │   │ │ Prefetch │ │      │
│  │ │   = 4    │ │   │ │   = 4    │ │   │ │   = 4    │ │      │
│  │ └──────────┘ │   │ └──────────┘ │   │ └──────────┘ │      │
│  └──────┬───────┘   └──────┬───────┘   └──────┬───────┘      │
│         │                  │                  │                │
│         └──────────────────┼──────────────────┘                │
│                            ▼                                   │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                   RESULT BACKEND                        │   │
│  │         (Redis / PostgreSQL / S3 / MongoDB)            │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                      MONITORING DASHBOARD                       │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Task Stats   │  │ Worker Stats │  │ Queue Depth  │          │
│  │ Real-time    │  │ Health       │  │ Chart        │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Success Rate │  │ Latency      │  │ Failed Tasks │          │
│  │ Graph        │  │ Histogram    │  │ Retry Queue  │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘

🚀 Quick Start

Installation

pip install taskqueue-pro

Basic Usage

from taskqueue import TaskQueue, task

# Create task queue
tq = TaskQueue(broker_url="redis://localhost:6379/0")

# Define a task
@task(queue="default", max_retries=3)
def send_email(to, subject, body):
    """Send email task"""
    import smtplib
    # Email sending logic
    print(f"Sending email to {to}")
    return {"status": "sent", "to": to}

# Enqueue task
result = send_email.delay("user@example.com", "Hello", "World!")
print(f"Task ID: {result.id}")

# Get result
print(result.get(timeout=10))

Running Workers

# Start worker
taskqueue worker --queues default,high --concurrency 4

# Start with auto-reload
taskqueue worker --queues default --autoreload

# Start scheduler
taskqueue scheduler --config scheduler.yml

Scheduling Tasks

from taskqueue import crontab

# Schedule task to run every hour
@task(queue="reports")
def generate_daily_report():
    """Generate daily report"""
    print("Generating report...")
    return {"status": "completed"}

# Schedule using cron syntax
generate_daily_report.schedule(crontab(hour="*/1"))

# Or use human-readable syntax
generate_daily_report.every("1 hour")

📊 Advanced Features

Task Chains

from taskqueue import chain

# Chain tasks
workflow = chain(
    download_file.s("https://example.com/data.csv"),
    process_data.s(),
    send_notification.s()
)

result = workflow.delay()

Task Groups

from taskqueue import group

# Run tasks in parallel
tasks = group([
    send_email.s(f"user{i}@example.com", "Hello", "Body")
    for i in range(10)
])

result = tasks.delay()
results = result.get()  # Get all results

Rate Limiting

@task(queue="api", rate_limit="10/m")  # 10 per minute
def call_external_api(endpoint):
    """Rate limited API call"""
    import requests
    return requests.get(endpoint).json()

Task Routing

# Route specific tasks to specific queues
@task(queue="gpu", routing_key="ml.*")
def train_model(dataset):
    """ML training task - routed to GPU workers"""
    pass

@task(queue="io", routing_key="file.*")
def process_large_file(filepath):
    """File processing task - routed to IO workers"""
    pass

🐳 Docker Deployment

# Clone repository
git clone https://github.com/efealtiparmakoglu/taskqueue-pro.git
cd taskqueue-pro

# Start infrastructure
docker-compose up -d redis postgres

# Start workers
docker-compose up -d worker

# Start API server
docker-compose up -d api

# Access dashboard
open http://localhost:8080

📈 Monitoring

Web Dashboard

# Start dashboard
taskqueue dashboard --port 8080

Prometheus Metrics

from taskqueue.metrics import task_completed, task_failed

# Custom metrics
@task()
def my_task():
    try:
        # Task logic
        task_completed.inc()
    except Exception:
        task_failed.inc()
        raise

Health Checks

# Check worker health
curl http://localhost:8080/health

# Get queue stats
curl http://localhost:8080/api/queues

# Get worker stats
curl http://localhost:8080/api/workers

🧪 Testing

# Test tasks synchronously
result = send_email.run("test@example.com", "Subject", "Body")

# Mock tasks for testing
from taskqueue.testing import task_mock

@task_mock
def test_task():
    assert send_email.delay().get() == {"status": "sent"}

📝 Configuration

YAML Config

# config/taskqueue.yml
broker:
  url: redis://localhost:6379/0
  
result_backend:
  url: postgresql://user:pass@localhost/taskqueue
  
worker:
  concurrency: 4
  prefetch_multiplier: 4
  task_time_limit: 3600
  max_tasks_per_child: 1000
  
queues:
  - name: default
    priority: 5
  - name: high
    priority: 10
  - name: low
    priority: 1
    
monitoring:
  enabled: true
  port: 8080
  metrics:
    enabled: true
    port: 9090

🛡️ Production Checklist

• Use Redis with persistence enabled
• Configure task result expiration
• Set up monitoring and alerts
• Enable task rate limiting
• Configure dead letter queues
• Set up log aggregation
• Enable SSL/TLS for broker connections
• Configure worker auto-scaling
• Set up backup for result backend

📚 API Reference

Task Decorator Options

@task(
    queue="default",           # Queue name
    priority=5,                # Task priority (1-10)
    max_retries=3,             # Max retry attempts
    retry_delay=60,            # Initial retry delay (seconds)
    retry_backoff=True,        # Use exponential backoff
    time_limit=3600,           # Hard time limit
    soft_time_limit=3000,      # Soft time limit
    rate_limit="100/h",        # Rate limit
    result_expires=86400,      # Result expiration (seconds)
    bind=True,                 # Pass self as first arg
    ignore_result=False,       # Don't store result
    store_errors_even_if_ignored=True
)

👥 Authors

• Efe Altıparmakoğlu - @efealtiparmakoglu

📄 License

MIT License - see LICENSE file