Metadata-Version: 2.4
Name: queue-max
Version: 0.1.0
Summary: Task queue with SQLite sharding, rate limiting, and circuit breaker
Author: Alexandre All
License: MIT
Project-URL: homepage, https://github.com/all451/queue-max
Project-URL: repository, https://github.com/all451/queue-max
Keywords: queue,task-queue,sqlite,background-tasks,worker,sharding
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Operating System :: OS Independent
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: typing-extensions>=4.5.0
Provides-Extra: webhook
Requires-Dist: requests>=2.31.0; extra == "webhook"
Provides-Extra: django
Requires-Dist: Django>=3.2; extra == "django"
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100.0; extra == "fastapi"
Provides-Extra: flask
Requires-Dist: Flask>=2.0; extra == "flask"
Provides-Extra: all
Requires-Dist: requests>=2.31.0; extra == "all"
Requires-Dist: Django>=3.2; extra == "all"
Requires-Dist: fastapi>=0.100.0; extra == "all"
Requires-Dist: Flask>=2.0; extra == "all"
Dynamic: license-file

# Queue Max

Task queue library with SQLite persistence, sharding, rate limiting, and circuit breaker.

No Redis or RabbitMQ required. Zero external dependencies (except typing-extensions).

## Installation

```bash
pip install queue-max
```

With framework integrations:

```bash
pip install queue-max[django]
pip install queue-max[fastapi]
pip install queue-max[flask]
```

## Quick Start

```python
from queue_max import Queue, Worker

# Create queue
queue = Queue()

# Enqueue a job
queue.enqueue({"task": "send_email", "to": "user@example.com"}, priority=2)

# Define processor
def process(payload):
    print(f"Processing: {payload}")

# Start worker
worker = Worker("worker-1", process, queue)
worker.start()
```

## Features

**SQLite Persistence** -- Jobs are stored in SQLite with WAL mode. No external services needed.

**Physical Sharding** -- Multiple .db files allow concurrent read/write across workers.

**Rate Limiting** -- Token bucket algorithm shared across all workers. Configurable per minute.

**Circuit Breaker** -- Stops calling failing services after N consecutive errors. Recovers automatically.

**Retry with Backoff** -- Exponential backoff with jitter for failed jobs. Configurable max retries.

**Heartbeat and Recovery** -- Workers send periodic heartbeats. Orphaned jobs are recovered automatically.

**Priority Queues** -- Three levels: low (0), medium (1), high (2).

**CLI** -- Built-in command line for stats, workers, and queue management.

## Configuration via Environment Variables

| Variable | Default | Description |
|----------|---------|-------------|
| NUM_SHARDS | 6 | Number of shard databases |
| RATE_LIMIT_MAX | 160 | Max requests per minute |
| QUEUE_MAX_RETRIES | 3 | Default max retry attempts |
| DB_BUSY_TIMEOUT | 30000 | SQLite busy timeout (ms) |
| HEARTBEAT_INTERVAL | 5000 | Worker heartbeat interval (ms) |
| STUCK_TIMEOUT | 30000 | Orphan job timeout (ms) |
| DATA_DIR | ./data | Directory for shard files |

## API Overview

### Queue

```python
from queue_max import Queue

queue = Queue(shards=6, rate_limit=160, max_retries=3)

# Enqueue jobs
queue.enqueue(payload, pagina_id=None, priority=0, max_retries=None)
queue.enqueue_batch([{"payload": {...}}, ...])

# Process jobs
job = queue.pop_job(worker_id)
queue.complete_job(job_id, shard_id)
queue.fail_job(job_id, shard_id, error, permanent=False)

# Management
queue.retry_failed_jobs()
queue.cleanup_old_jobs(days=7)
queue.recover_orphans()
stats = queue.get_stats()
```

### Worker

```python
from queue_max import Worker, WorkerPool

worker = Worker("worker-1", process_function, queue)
worker.start()
worker.stop()

pool = WorkerPool([worker1, worker2])
pool.start_all()
pool.stop_all()
```

### Decorator

```python
from queue_max import task

@task(priority=2, max_retries=3)
def send_email(to: str, subject: str):
    return send(to, subject)

send_email.delay("user@example.com", "Hello")
```

### CLI

```bash
queue-max stats
queue-max worker --function mymodule:myfunction --workers 4
queue-max enqueue --payload '{"task": "test"}' --priority 2
queue-max list --status failed --limit 20
queue-max retry
queue-max purge --days 7
```

## Framework Integrations

### Django

```python
# settings.py
INSTALLED_APPS = ["queue_max.contrib.django", ...]
QUEUE_MAX = {"SHARDS": 4, "RATE_LIMIT": 160}

# tasks.py
from queue_max.contrib.django import task
@task
def my_task(user_id): ...
```

Management commands: `python manage.py queue_worker`, `queue_stats`, `queue_purge`.

### FastAPI

```python
from fastapi import FastAPI
from queue_max.contrib.fastapi import QueueMiddleware

app = FastAPI()
app.add_middleware(QueueMiddleware, max_workers=4)
```

### Flask

```python
from flask import Flask
from queue_max.contrib.flask import QueueExtension

app = Flask(__name__)
queue = QueueExtension(app)

@queue.task
def my_task(): ...
```

## Performance

| Cenário | Throughput |
|---------|-----------|
| Burst (20 workers, 10 shards) | **~3.300 jobs/sec** |
| Contenção (10 workers, 1 shard) | **~1.660 jobs/sec** |
| Com 30% de falhas (8 workers) | **Estável** — circuit breaker não trip |

- Max queue size: 1M+ jobs per shard
- [Resultados detalhados dos stress tests](docs/stress-test.md)

## Running Tests

```bash
pip install -e ".[dev]"
pytest tests/ -v
```

## License

MIT
