DOCUMENTATION

INSTALLATION

pip install openqueue-pg

INITIALIZATION

from openqueue import OpenQueue

client = OpenQueue(
    base_url="https://your-api-url.com",
    api_token="oq_live_..."
)

PRODUCER - ENQUEUE JOBS

# Simple job
job_id = client.enqueue(
    queue_name="emails",
    payload={"to": "user@example.com", "subject": "Hello"}
)

# Scheduled job (run later)
job_id = client.enqueue(
    queue_name="reminders",
    payload={"user_id": 123, "message": "Reminder!"},
    run_at="2026-01-01T09:00:00Z"
)

# With priority (higher = more urgent)
job_id = client.enqueue(
    queue_name="emails",
    payload={"to": "user@example.com"},
    priority=10
)

# Batch enqueue
job_ids = client.enqueue_batch([
    {"queue_name": "emails", "payload": {"to": "a@b.com"}},
    {"queue_name": "emails", "payload": {"to": "c@d.com"}, "priority": 10},
])

WORKER - PROCESS JOBS

while True:
    leased = client.lease(queue_name="emails", worker_id="worker-1")
    
    if leased:
        try:
            # Process the job
            payload = leased.job.payload
            print(f"Processing: {payload}")
            
            # Success - mark complete
            client.ack(
                leased.job.id,
                leased.lease_token,
                result={"done": True}
            )
        except Exception as e:
            # Failure - retry
            client.nack(
                leased.job.id,
                leased.lease_token,
                error=str(e)
            )

HEARTBEAT

For long-running jobs, extend the lease to prevent timeout:

# Extend lease by 30 seconds
client.heartbeat(
    job_id="...",
    lease_token="...",
    lease_seconds=30
)

JOB OPERATIONS

# Get job status
status = client.get_status(job_id="...")

# Get full job details
job = client.get_job(job_id="...")

# List jobs
jobs = client.list_jobs(
    queue_name="emails",
    status="completed",
    limit=50,
    offset=0
)

# Cancel pending job
client.cancel_job(job_id="...")

# Get queue statistics
stats = client.get_queue_stats()

EXCEPTIONS