zerodb-rq

Python RQ-compatible task queue backed by ZeroDB. No Redis required -- auto-provisions on first use.

Why zerodb-rq?

	Python RQ	Celery	zerodb-rq
Requires	Redis	Redis/RabbitMQ	Nothing
Setup	Install + configure Redis	Install + configure broker	pip install zerodb-rq
Persistence	Redis (volatile)	Configurable	ZeroDB (cloud, durable)
Auto-provisioning	No	No	Yes (free, no signup)

Installation

pip install zerodb-rq

Quick Start

Enqueue Jobs

from zerodb_rq import Queue

def send_email(to, subject, body):
    print(f"Sending email to {to}: {subject}")

q = Queue()  # auto-provisions ZeroDB
job = q.enqueue(send_email, "alice@example.com", "Hello", body="Hi there")
print(f"Job {job.id} enqueued")

Run a Worker

from zerodb_rq import Queue, Worker

q = Queue()
w = Worker([q])
w.work()  # Polls and executes jobs

Burst Mode (Run Once)

w = Worker([q])
w.work(burst=True)  # Process all queued jobs, then stop

How It Works

1. Queue.enqueue() serializes the function call and stores it in a ZeroDB table (rq_jobs)
2. Worker.work() polls the table for queued jobs, dequeues them FIFO, and executes
3. Results are stored in a separate ZeroDB table (rq_results)
4. On first use, ZeroDB is auto-provisioned (free, no signup)

API Reference

Queue(name="default", api_key=None, project_id=None)

Method	Description
enqueue(func, *args, **kwargs)	Add a job to the queue
enqueue_call(func, args, kwargs)	Add a job with explicit args
count()	Number of queued jobs
is_empty	Whether the queue is empty
empty()	Remove all queued jobs
fetch_job(job_id)	Get a specific job
get_jobs()	List all queued jobs
dequeue()	Pop the next job
fetch_result(job_id)	Get a completed job's result

Worker(queues, poll_interval=1)

Method	Description
work(burst=False, max_jobs=None)	Start processing jobs
stop()	Signal graceful shutdown
is_running	Whether the worker is active
total_completed	Jobs completed count
total_failed	Jobs failed count

Job

Attribute	Description
id	Unique job identifier
func_name	Dotted function path
args	Positional arguments
kwargs	Keyword arguments
status	queued / started / finished / failed
result	Return value (after execution)
error	Error message (if failed)

Configuration

Explicit Credentials

q = Queue(api_key="your-api-key", project_id="your-project-id")

Environment Variables

export ZERODB_API_KEY="your-api-key"
export ZERODB_PROJECT_ID="your-project-id"

Credential Resolution Order

1. Constructor arguments
2. Environment variables (ZERODB_API_KEY, ZERODB_PROJECT_ID)
3. Config file (~/.zerodb/config.json)
4. Auto-provision (free, no signup)

Examples

Multiple Queues

high = Queue(name="high-priority")
low = Queue(name="low-priority")

high.enqueue(critical_task, data)
low.enqueue(background_task, data)

# Worker processes high-priority first
w = Worker([high, low])
w.work()

Custom Job ID

job = q.enqueue(process_order, order_id, job_id=f"order-{order_id}")

Job Results

job = q.enqueue(calculate, 42)
# ... later, after worker processes it ...
result = q.fetch_result(job.id)

License

MIT

---

Powered by ZeroDB + AINative

ZeroDB is a serverless database with auto-provisioning, vector search, and NoSQL tables. Build AI-native apps without managing infrastructure.

• Documentation
• GitHub
• Discord