Spooled Python SDK

Official Python SDK for Spooled Cloud — a modern, scalable job queue and task scheduler.

Live Demo (SpriteForge) • Documentation

Features

• Full API Coverage — Access all Spooled API endpoints
• Type Safety — Full type hints and Pydantic models
• Sync & Async — Both synchronous and asynchronous clients
• Worker Runtime — Process jobs with a decorator-based API
• Real-time Events — WebSocket and SSE support
• gRPC Support — High-performance gRPC client (optional)
• Resilience — Retry logic with exponential backoff and circuit breaker
• Production Ready — Comprehensive error handling and logging

Installation

pip install spooled

Optional Extras

# For real-time events (WebSocket/SSE)
pip install spooled[realtime]

# For gRPC support
pip install spooled[grpc]

# All features
pip install spooled[all]

Quick Start

Create a Job

from spooled import SpooledClient

client = SpooledClient(api_key="sp_live_...")

# Create a job
result = client.jobs.create({
    "queue_name": "emails",
    "payload": {
        "to": "user@example.com",
        "subject": "Welcome!",
        "template": "welcome_email",
    },
    "priority": 5,
})

print(f"Job created: {result.id}")

# Get job details
job = client.jobs.get(result.id)
print(f"Status: {job.status}")

client.close()

Process Jobs with a Worker

from spooled import SpooledClient
from spooled.worker import SpooledWorker

client = SpooledClient(api_key="sp_live_...")
worker = SpooledWorker(client, queue_name="emails", concurrency=10)

@worker.process
def handle_job(ctx):
    """Process an email job."""
    print(f"Processing job {ctx.job_id}")
    
    # Access payload
    to = ctx.payload["to"]
    subject = ctx.payload["subject"]
    
    # Send the email (your logic here)
    send_email(to, subject)
    
    # Return result
    return {"sent": True}

@worker.on("job:completed")
def on_completed(event):
    print(f"Job {event.job_id} completed!")

worker.start()  # Blocking

Real-world examples (beginner friendly)

If you want 5 copy/paste “real life” setups (Stripe → jobs, GitHub Actions → jobs, cron schedules, CSV import, website signup), see:

• https://github.com/spooled-cloud/spooled-backend/blob/main/docs/guides/real-world-examples.md

Async Client

import asyncio
from spooled import AsyncSpooledClient

async def main():
    async with AsyncSpooledClient(api_key="sp_live_...") as client:
        # Create multiple jobs concurrently
        tasks = [
            client.jobs.create({"queue_name": "tasks", "payload": {"n": i}})
            for i in range(10)
        ]
        results = await asyncio.gather(*tasks)
        print(f"Created {len(results)} jobs")

asyncio.run(main())

Workflows (Job Dependencies)

from spooled import SpooledClient

client = SpooledClient(api_key="sp_live_...")

# Create a workflow with dependencies
workflow = client.workflows.create({
    "name": "Order Processing",
    "jobs": [
        {
            "key": "validate",
            "queue_name": "orders",
            "payload": {"action": "validate"},
        },
        {
            "key": "charge",
            "queue_name": "payments",
            "payload": {"action": "charge"},
            "depends_on": ["validate"],
        },
        {
            "key": "ship",
            "queue_name": "fulfillment",
            "payload": {"action": "ship"},
            "depends_on": ["charge"],
        },
    ],
})

print(f"Workflow created: {workflow.workflow_id}")

Schedules (Cron Jobs)

from spooled import SpooledClient

client = SpooledClient(api_key="sp_live_...")

# Create a scheduled job
schedule = client.schedules.create({
    "name": "Daily Report",
    "cron_expression": "0 9 * * *",  # 9 AM daily
    "timezone": "America/New_York",
    "queue_name": "reports",
    "payload_template": {"report_type": "daily"},
})

print(f"Schedule created: {schedule.id}")
print(f"Next run: {schedule.next_run_at}")

gRPC Support (High Performance)

The SDK includes a high-performance gRPC client for high-throughput worker scenarios.

from spooled.grpc import SpooledGrpcClient

# Connect to Spooled Cloud gRPC (TLS required for Cloudflare Tunnel)
client = SpooledGrpcClient(
    address="grpc.spooled.cloud:443", 
    api_key="sp_live_...",
    use_tls=True  # Required for production (Cloudflare Tunnel needs HTTPS for HTTP/2)
)

# Enqueue a job
response = client.queue.enqueue(
    queue_name="high-throughput",
    payload={"data": "value"}
)
print(f"Job enqueued: {response.job_id}")

# Get queue stats
stats = client.queue.get_queue_stats("high-throughput")
print(f"Pending jobs: {stats.pending}")

client.close()

When to use gRPC?

• High Throughput: 3x faster than HTTP API for enqueue/dequeue operations.
• Streaming: Supports real-time job streaming.
• Efficiency: Uses persistent HTTP/2 connections with keepalives.

Real-time Events (SSE/WebSocket)

Listen for real-time updates:

from spooled import SpooledClient
from spooled.realtime.unified import SubscriptionFilter

client = SpooledClient(api_key="sp_live_...")

# Subscribe to job updates for a queue, then iterate events
realtime = client.realtime(type="websocket")
realtime.connect()
realtime.subscribe(SubscriptionFilter(queue="my-queue"))

for event in realtime.events():
    print(f"Event: {event.type} - {event.data}")

Configuration

from spooled import SpooledClient, SpooledClientConfig, RetryConfig

# Full configuration
config = SpooledClientConfig(
    api_key="sp_live_...",
    base_url="https://api.spooled.cloud",
    timeout=30.0,
    retry=RetryConfig(
        max_retries=3,
        base_delay=1.0,
        max_delay=30.0,
    ),
    debug=True,
)

client = SpooledClient(config=config)

Error Handling

from spooled import SpooledClient
from spooled.errors import (
    SpooledError,
    NotFoundError,
    ValidationError,
    RateLimitError,
)

client = SpooledClient(api_key="sp_live_...")

try:
    job = client.jobs.get("nonexistent")
except NotFoundError:
    print("Job not found")
except RateLimitError as e:
    # 429 covers both per-second rate limiting and plan-quota errors.
    # A plan quota carries code == "QUOTA_EXCEEDED" (with resource/current/limit/plan details).
    if e.code == "QUOTA_EXCEEDED":
        print(f"Plan quota exceeded: {e.message}")
    else:
        print(f"Rate limited. Retry after {e.retry_after}s")
except SpooledError as e:
    print(f"API error: {e.code} - {e.message}")

API Reference

Client Resources

Worker Events

Environment Variables

Requirements

• Python 3.10+
• httpx>=0.25.0
• pydantic>=2.0.0

License

Apache 2.0