BaQueue

A powerful Python queue management package. Multi-driver support, batch jobs, scheduling, auto-balancing, and a beautiful real-time monitoring dashboard.

Jobs - Filterable job list with status, attempts, and duration

Queues - Per-queue detail cards with pending/processing/completed/failed

Table of Contents

• Features
• Quick Start
  • Define a Job
  • Dispatch Jobs
  • Batch Jobs
  • Run Workers
    • Stuck Job Recovery
  • Pruning
    • Auto-pruning
    • Manual pruning
  • Retry Failed Jobs
  • Dashboard
    • Workers Tab (Supervisor/Worker Monitoring)
  • Drivers
• Examples
• Testing
• CLI Commands
• Benchmarks
  • Test 1: 1,000 jobs / 5 workers
  • Test 2: 5,000 jobs / 10 workers
  • Stress Test Options
  • Run with Live Dashboard
• License

Features

• Multi-driver: SQLite (default), Redis, PostgreSQL, or In-Memory
• Auto-balancing: Dynamically scale workers based on queue pressure
• Stuck-job recovery: Jobs left in processing for more than 1 hour are requeued automatically
• Auto-pruning: Completed jobs are deleted about 5 seconds after they finish; failed/cancelled jobs are kept up to 1 day — all configurable
• Disk-full cleanup: Storage-full/OOM driver errors trigger emergency cleanup of terminal jobs and old metrics, then retry once
• Pruning: Remove old jobs by status, tag, or age
• Monitoring Dashboard: Real-time WebSocket-powered UI with date filtering
• CLI: Manage workers, scheduler, dashboard, and pruning from the command line
• Cross-process: SQLite driver shares state between dashboard and workers without external dependencies

Quick Start

# Install (SQLite + in-memory drivers work out of the box, zero extra dependencies)
pip install baqueue

# With Redis support
pip install "baqueue[redis]"

# With PostgreSQL support
pip install "baqueue[postgres]"

# With dashboard
pip install "baqueue[dashboard]"

# Everything
pip install "baqueue[all]"

Installing from source (development):

git clone https://github.com/basalam/baqueue.git
cd baqueue
pip install -e ".[dev]"   # editable install with all extras + test/build tooling

Define a Job

from baqueue import Job

class SendEmail(Job):
    queue = "emails"
    max_attempts = 3
    backoff = "exponential"

    async def handle(self, to: str, subject: str, body: str):
        await send_email(to, subject, body)

    async def on_failure(self, error, payload):
        print(f"Failed to send email: {error}")

Or use the decorator:

from baqueue import Job

@Job.as_job(queue="emails", max_attempts=3)
async def send_email(to, subject, body):
    ...

Dispatch Jobs

from baqueue import Queue, BaQueueConfig
from baqueue.config import DriverConfig

# Configure (SQLite driver by default - works across processes)
Queue.configure(BaQueueConfig(driver=DriverConfig(name="sqlite")))
await Queue.connect()

# Push a job
await Queue.push(SendEmail, to="user@example.com", subject="Hi", body="Hello!")

# Push with delay (60 seconds)
await Queue.later(SendEmail, delay=60, to="user@example.com", subject="Reminder", body="...")

# Bulk push (much faster for large volumes)
await Queue.bulk([
    (SendEmail, {"to": "a@b.com", "subject": "Hi", "body": "A"}),
    (SendEmail, {"to": "c@d.com", "subject": "Hi", "body": "B"}),
])

Batch Jobs

from baqueue import Batch

result = await Batch(driver, [
    (SendEmail, {"to": "a@b.com", "subject": "Hi", "body": "Hey"}),
    (SendEmail, {"to": "c@d.com", "subject": "Hi", "body": "Hey"}),
]).name("newsletter").then(OnAllDone).catch(OnAnyFailed).dispatch()

Run Workers

from baqueue.supervisor import Supervisor
from baqueue.config import SupervisorConfig

supervisor = Supervisor(
    driver=Queue.get_driver(),
    config=SupervisorConfig(
        queues=["emails", "payments"],
        min_workers=3,
        max_workers=10,
        balance="auto",
    ),
)
await supervisor.start()

Or via CLI:

baqueue work -q emails -q payments -w 3 -b auto

Stuck Job Recovery

When baqueue work is running, the supervisor also checks for jobs that were claimed by a worker but never finished. By default, any job that has stayed in processing for more than 1 hour is moved back to pending, so another worker can pick it up and run it again.

This is intended for worker crashes, process restarts, or other cases where a job was left in-flight. The original claim still counts as an attempt; when the job is picked up again, its attempt counter continues from there.

Configure it from Python:

supervisor = Supervisor(
    driver=Queue.get_driver(),
    config=SupervisorConfig(
        queues=["emails"],
        recover_stuck_jobs=True,
        stuck_processing_seconds=3600,
        stuck_check_interval_seconds=60,
    ),
)

Or from the CLI:

baqueue work --stuck-job-timeout-seconds 7200
baqueue work --no-stuck-job-recovery

If you intentionally run jobs for longer than 1 hour, increase stuck_processing_seconds above your longest expected runtime.

Pruning

Auto-pruning (runs alongside baqueue work)

When baqueue work is running, a background pruner cycles every 5s and applies these defaults across every driver:

Override from a JSON config file (baqueue -c config.json work):

{
  "auto_prune": true,
  "prune_completed_seconds": 600,
  "prune_other_seconds": 172800,
  "prune_interval_seconds": 30,
  "auto_cleanup_on_disk_full": true
}

Or from Python:

from baqueue import BaQueueConfig
config = BaQueueConfig(
    prune_completed_seconds=600,    # 10 minutes
    prune_other_seconds=172800,     # 2 days
    prune_interval_seconds=30,
    auto_cleanup_on_disk_full=True, # enabled by default
)

Or from the CLI:

baqueue work --prune-completed-seconds 600 --prune-other-seconds 172800
baqueue work --no-auto-prune              # disable the background pruner
baqueue work --no-disk-full-cleanup       # disable emergency storage cleanup

The legacy hour-based fields (prune_completed_hours, prune_failed_hours, prune_cancelled_hours, prune_metrics_hours) are still respected for backward compatibility — when set to a positive value they override the corresponding *_seconds field.

Disk-full emergency cleanup

auto_cleanup_on_disk_full is enabled by default. When a driver write/update/delete operation sees a storage-full style error (SQLite disk full, PostgreSQL disk/memory exhausted, Redis OOM/maxmemory), BaQueue runs an emergency cleanup that removes terminal jobs (completed, failed, cancelled) and old metrics, then retries the failed operation once. If cleanup does not free enough space, the original driver error is still raised.

Manual pruning

# Remove completed jobs older than 24 hours
await Queue.prune(status="completed", hours=24)

# Remove jobs by tag
await Queue.prune(tag="batch:newsletter")

Redis index health

The Redis driver keeps secondary indexes (sorted sets) so the dashboard can list and count jobs by queue/status efficiently. All deletes go through an index-consistent path that removes the job hash and every index entry in one atomic step, so the indexes stay bounded. If entries are ever orphaned out-of-band (e.g. job hashes deleted directly via redis-cli), pruning reaps them automatically, and you can force a full repair:

baqueue reconcile-indexes -d redis --driver-url redis://localhost:6379/0

Set reconcile_on_connect=True to run that repair once on every startup (off by default to keep connect fast on large datasets).

If a Redis failure occurred during the old non-atomic ready-list dequeue path, repair pending jobs that are present in the pending index but missing from both the ready list and delayed set:

baqueue reconcile-pending -d redis --driver-url redis://localhost:6379/0

The repair is safe to repeat. Each job is handled in its own Redis transaction: future jobs return to the delayed sorted set, runnable jobs get exactly one ready-list entry, and missing hashes lose stale queue/index references.

Retry Failed Jobs

Bulk-retry failed jobs from the CLI, from Python, or from the dashboard.

CLI:

# Retry every failed job (asks for confirmation)
baqueue retry-failed

# Skip the confirmation prompt
baqueue retry-failed -y

# Limit to a specific queue
baqueue retry-failed -q emails

# Combine filters: queue + tag + age window
baqueue retry-failed -q emails -t campaign --hours 24

# Use a non-default driver
baqueue retry-failed -d redis --driver-url redis://localhost:6379/0

Python:

# Retry every failed job
count = await Queue.retry_failed()

# Retry only failed jobs in a queue
count = await Queue.retry_failed(queue="emails")

# Filter by tag and creation window
from baqueue.serializer import _now_ts
count = await Queue.retry_failed(
    queue="emails",
    tag="campaign",
    created_from=_now_ts() - 24 * 3600,
)

Dashboard: open the Jobs tab, set the Status filter to Failed, then click the amber Retry All button. The current Queue / Tag / date-range filters are respected.

Each matched job is released back onto its queue with delay=0, the same path used by single-job retry.

Dashboard

# Start the dashboard (uses SQLite by default)
baqueue dashboard

# Open http://localhost:9100

The dashboard includes:

• Real-time overview with pending/processing/completed/failed counters sourced from live job state (not bounded metric logs)
• Date range filtering (custom range + presets: 1h, 24h, 7d, 30d)
• Job detail modal with timeline, payload data, and error trace
• Queue breakdown with progress bars
• Worker monitoring with active/idle status
• Dark/light theme toggle
• Scheduled-job badge with hover tooltip showing exact execution time, plus a "Scheduled For" entry in the job timeline
• Bulk "Retry All" button when the Jobs view is filtered to failed (respects the active queue/tag/date filters)
• Queue filter as a dropdown auto-populated from active queues (no manual typing)
• Mobile-friendly sidebar drawer with hamburger toggle on screens ≤900px

Run in one terminal:

baqueue dashboard

Dispatch jobs in another terminal:

python examples/simple_job.py

Workers Tab (Supervisor/Worker Monitoring)

To see active supervisors/workers in the Workers tab, work and dashboard must point to the same backend (same driver and same URL/path).

Example with SQLite:

Terminal 1:

baqueue work -d sqlite --driver-url .baqueue.db -q default -w 3

Terminal 2:

baqueue dashboard -d sqlite --driver-url .baqueue.db

Then open:

http://localhost:9100

Quick troubleshooting:

• Check http://localhost:9100/api/supervisors (should return a non-empty supervisors list while workers are running).
• If api/supervisors is empty, work and dashboard are likely using different driver URLs/paths.
• memory driver is single-process only, so separate work and dashboard processes will not share worker state.

Driver-specific CLI examples:

SQLite (shared local file):

baqueue work -d sqlite --driver-url .baqueue.db -q default -w 3
baqueue dashboard -d sqlite --driver-url .baqueue.db

Redis (shared Redis DB):

baqueue work -d redis --driver-url redis://localhost:6379/0 -q default -w 3
baqueue dashboard -d redis --driver-url redis://localhost:6379/0

PostgreSQL (shared database/schema):

baqueue work -d postgres --driver-url postgresql://user:pass@localhost/dbname -q default -w 3
baqueue dashboard -d postgres --driver-url postgresql://user:pass@localhost/dbname

Memory (single-process only):

# Use an in-process example to run workers + dashboard together.
python examples/dashboard_demo.py

Drivers

SQLite (default, zero-config, cross-process):

Queue.configure(BaQueueConfig(
    driver=DriverConfig(name="sqlite")
))

Redis:

Queue.configure(BaQueueConfig(
    driver=DriverConfig(name="redis", url="redis://localhost:6379/0")
))

PostgreSQL:

Queue.configure(BaQueueConfig(
    driver=DriverConfig(name="postgres", url="postgresql://user:pass@localhost/dbname")
))

Memory (single-process testing only):

Queue.configure(BaQueueConfig(
    driver=DriverConfig(name="memory")
))

Examples

# Simple job processing
python examples/simple_job.py

# Batch processing
python examples/batch_example.py

# Scheduled jobs
python examples/scheduled_example.py

# Dashboard demo (open http://localhost:9100)
python examples/dashboard_demo.py

# Delayed jobs demo — shows the "Scheduled" badge with varied delays
python examples/delayed_jobs_demo.py

# Stress test (see Benchmarks section below)
python examples/stress_test.py --jobs 1000 --workers 5 --bulk

Testing

The full test suite lives in tests/ and runs with one command:

# Run everything
baqueue test

# Quiet output, stop at the first failure
baqueue test -q -x

# Run only retry-failed related tests
baqueue test -k "RetryFailed or retry_failed"

# Re-run just the tests that failed last time
baqueue test --last-failed

# Filter by marker (markers defined in pyproject.toml)
baqueue test -m "not slow"

baqueue test is a thin wrapper around pytest, so it picks up the project's tool.pytest.ini_options config (asyncio mode, marker definitions, etc.). You can also run pytest directly:

pip install baqueue[dev]
pytest tests/ -v

Coverage includes:

• Serializer / payload roundtrip (incl. delay_until)
• Backoff strategies (fixed, linear, exponential, explicit list)
• Job + FunctionJob + @Job.as_job decorator
• Queue facade — push / later / bulk / prune / retry_failed
• Cross-driver contract tests (memory + sqlite, parameterized)
• Worker lifecycle: success / failure / retry / timeout
• Supervisor pool + delayed-job promotion + stuck-job recovery
• Scheduler interval dispatch
• Pruner by status / tag / age
• Batch builder + completion callbacks
• DashboardAPI (overview, jobs_list, retry, bulk retry-failed, prune, stats)
• CLI command surface (help text, validation, retry-failed abort flow)
• Redis claim atomicity, concurrent workers, delayed recovery, pending reconciliation, and secondary-index consistency

CLI Commands

baqueue work          Start processing jobs
baqueue schedule      Start the job scheduler
baqueue dashboard     Launch the monitoring dashboard
baqueue prune         Prune old jobs
baqueue retry-failed  Retry all failed jobs (filter by queue/tag/age)
baqueue reconcile-indexes  Repair Redis secondary indexes (drop stale entries)
baqueue reconcile-pending  Repair pending jobs missing from ready/delayed storage
baqueue status        Show queue status
baqueue test          Run the test suite

Use -h on any command for options:

baqueue -h
baqueue work -h
baqueue dashboard -h

Benchmarks

Stress tests run on Windows 10, Python 3.11, SQLite driver, using examples/stress_test.py.

The stress test dispatches jobs across 5 queues (fast, slow, flaky, heavy, notifications) with varying execution times and a ~30% failure rate on the flaky queue, exercising retries and backoff.

Test 1: 1,000 jobs / 5 workers

python examples/stress_test.py --jobs 1000 --workers 5 --bulk

============================================================
  RESULTS
============================================================
  Total time:    30.38s
  Completed:     993
  Failed:        7
  Throughput:    32.9 jobs/s
  Success rate:  99.3%
============================================================

Test 2: 5,000 jobs / 10 workers

python examples/stress_test.py --jobs 5000 --workers 10 --bulk

============================================================
  RESULTS
============================================================
  Total time:    49.95s
  Completed:     4965
  Failed:        35
  Throughput:    100.1 jobs/s
  Success rate:  99.3%
============================================================

Stress Test Options

python examples/stress_test.py [OPTIONS]

Options:
  --jobs, -j      Number of jobs to dispatch (default: 1000)
  --workers, -w   Number of concurrent workers (default: 5)
  --bulk          Use bulk insert for faster dispatching
  --dashboard     Launch live dashboard on http://localhost:9100

Job types used in the stress test:

Run with Live Dashboard

python examples/stress_test.py --jobs 3000 --workers 8 --bulk --dashboard
# Open http://localhost:9100 to watch progress in real-time

License

MIT © Basalam and BaQueue Contributors