lilith-queue-cli

Python CLI tools for managing BullMQ queues via Redis.

Installation

pip install lilith-queue-cli

Or from the Lilith Platform registry:

pip install lilith-queue-cli --index-url http://forge.black.lan/api/packages/lilith/pypi/simple/

Commands

queue-status

View queue status and job counts.

queue-status -q IMAGE_GENERATOR_QUEUE
queue-status -q my-queue -r redis://custom:6379
queue-status -q my-queue -j 5  # Show 5 recent jobs

queue-list

List jobs in the queue.

queue-list -q my-queue -s waiting
queue-list -q my-queue -s failed -l 50
queue-list -q my-queue -s waiting -f "cyberpunk"  # Filter by name
queue-list -q my-queue -s waiting -v  # Verbose output

queue-clear

Clear jobs from the queue.

queue-clear -q my-queue --waiting
queue-clear -q my-queue --failed --completed
queue-clear -q my-queue --all
queue-clear -q my-queue --waiting -f "test"  # Filter by name
queue-clear -q my-queue --active --force  # Force clear active jobs
queue-clear -q my-queue --waiting --dry-run  # Preview without clearing

queue-control

Control queue operations.

queue-control -q my-queue pause
queue-control -q my-queue resume
queue-control -q my-queue drain  # Remove all waiting jobs

Environment Variables

• REDIS_URL: Default Redis connection URL (default: redis://localhost:6379)

Python API

from lilith_queue_cli.client import QueueClient, JobCounts, JobInfo

client = QueueClient("my-queue", redis_url="redis://localhost:6379")

# Get job counts
counts: JobCounts = client.get_job_counts()
print(f"Waiting: {counts.waiting}, Active: {counts.active}")

# Get jobs
jobs: list[JobInfo] = client.get_jobs(["waiting", "failed"], limit=100)
for job in jobs:
    print(f"[{job.state}] {job.name}")

# Control queue
client.pause()
client.resume()
client.drain()

# Clear jobs
cleared, errors = client.clear_jobs(["completed", "failed"])
print(f"Cleared {cleared} jobs")

client.close()

License

MIT