queueflow-sdk 0.1.0

Python SDK for Queueflow distributed task queue

QueueFlow SDK

Python SDK for the QueueFlow distributed task queue.

Installation

pip install queueflow-sdk

Quick Start

from queueflow_sdk import QueueFlowClient

# Create a tenant and get an API key
tenant = QueueFlowClient.create_tenant("https://your-server.com", "Company Name")
key = QueueFlowClient.create_api_key("https://your-server.com", tenant.id)

# Initialize the client
qf = QueueFlowClient("https://your-server.com", key.key)

# Submit a task
task = qf.submit("send_email", payload = {
    "to": "[e-mail]",
    "subject": "Hello from QueueFlow",
})

# Wait for completion
result = qf.wait_for(task.id)
print(result.status) # "completed"
print(result.max_results) # {"sent_to": "[e-mail]", ...}

# Always close when done
qf.close()

Context Manager

with QueueFlowClient("https://your-server.com", "your-api-key") as qf:
    task = qf.submit("process_image", payload = {
        "image_url": "https://example.com/photo.jpg",
        "width": 400,
        "height": 300,
    })
    result = qf.wait_for(task.id)

Error Handling

from queueflow_sdk import QueueFlowClient, ConflictError, NotFoundError

qf = QueueFlowClient("https://your-server.com", "your-api-key")

try:
    qf.cancel("nonexistent-task-id")
except NotFoundError:
    print("Task not found")
except ConflictError:
    print("Task already completed")