Examples

Common Compute API workflows with complete code examples.

---

Task Submission

Submit Generation Task

cURLPython

curl -X POST https://compute-api.sartiq.com/api/v1/tasks/generation \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "input_image_url": "https://storage.sartiq.com/products/dress-001.jpg",
    "prompt": "fashion model wearing the dress, studio lighting, white background",
    "negative_prompt": "blurry, distorted, low quality",
    "num_outputs": 4,
    "steps": 30,
    "guidance_scale": 7.5
  }'

import httpx

response = httpx.post(
    "https://compute-api.sartiq.com/api/v1/tasks/generation",
    headers={"Authorization": f"Bearer {api_key}"},
    json={
        "input_image_url": "https://storage.sartiq.com/products/dress-001.jpg",
        "prompt": "fashion model wearing the dress, studio lighting",
        "num_outputs": 4,
        "steps": 30
    }
)
task = response.json()
print(f"Task ID: {task['id']}")

Response:

{
  "id": "task_abc123",
  "type": "generation",
  "status": "pending",
  "created_at": "2024-01-15T10:30:00Z"
}

---

Polling for Completion

Check Task Status

curl https://compute-api.sartiq.com/api/v1/tasks/task_abc123 \
  -H "Authorization: Bearer $API_KEY"

Response (in progress):

{
  "id": "task_abc123",
  "status": "running",
  "progress": 0.65,
  "started_at": "2024-01-15T10:30:05Z"
}

Response (completed):

{
  "id": "task_abc123",
  "status": "completed",
  "progress": 1.0,
  "output": {
    "images": [
      {"url": "https://storage.sartiq.com/outputs/img1.jpg", "seed": 12345},
      {"url": "https://storage.sartiq.com/outputs/img2.jpg", "seed": 12346},
      {"url": "https://storage.sartiq.com/outputs/img3.jpg", "seed": 12347},
      {"url": "https://storage.sartiq.com/outputs/img4.jpg", "seed": 12348}
    ]
  },
  "completed_at": "2024-01-15T10:31:00Z"
}

Poll Until Complete

PythonTypeScript

import time
import httpx

def wait_for_task(task_id: str, api_key: str, timeout: int = 300) -> dict:
    """Poll until task completes or timeout."""
    base_url = "https://compute-api.sartiq.com/api/v1"
    headers = {"Authorization": f"Bearer {api_key}"}

    start = time.time()
    while time.time() - start < timeout:
        response = httpx.get(
            f"{base_url}/tasks/{task_id}",
            headers=headers
        )
        task = response.json()

        if task["status"] == "completed":
            return task
        elif task["status"] == "failed":
            raise Exception(f"Task failed: {task.get('error')}")
        elif task["status"] == "cancelled":
            raise Exception("Task was cancelled")

        # Progress update
        print(f"Progress: {task.get('progress', 0) * 100:.0f}%")
        time.sleep(2)

    raise TimeoutError("Task timed out")

# Usage
task = wait_for_task("task_abc123", api_key)
for image in task["output"]["images"]:
    print(image["url"])

async function waitForTask(
  taskId: string,
  api_key: string,
  timeout = 300000
): Promise<any> {
  const baseUrl = "https://compute-api.sartiq.com/api/v1";
  const headers = { Authorization: `Bearer ${api_key}` };
  const start = Date.now();

  while (Date.now() - start < timeout) {
    const response = await fetch(`${baseUrl}/tasks/${taskId}`, {
      headers,
    });
    const task = await response.json();

    if (task.status === "completed") {
      return task;
    } else if (task.status === "failed") {
      throw new Error(`Task failed: ${task.error}`);
    } else if (task.status === "cancelled") {
      throw new Error("Task was cancelled");
    }

    console.log(`Progress: ${(task.progress || 0) * 100}%`);
    await new Promise((r) => setTimeout(r, 2000));
  }

  throw new Error("Task timed out");
}

---

Training Jobs

Start Training

curl -X POST https://compute-api.sartiq.com/api/v1/training/ \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "dataset_url": "https://storage.sartiq.com/datasets/model-ref.zip",
    "base_model": "sd-xl-base",
    "output_name": "custom-model-v1",
    "config": {
      "epochs": 100,
      "learning_rate": 1e-4,
      "batch_size": 1
    }
  }'

Monitor Training

curl https://compute-api.sartiq.com/api/v1/training/train_xyz789 \
  -H "Authorization: Bearer $API_KEY"

{
  "id": "train_xyz789",
  "status": "running",
  "progress": 0.45,
  "output": {
    "metrics": {
      "current_epoch": 45,
      "loss": 0.023
    }
  }
}

---

Workflows

Create and Start Workflow

# Create workflow
curl -X POST https://compute-api.sartiq.com/api/v1/workflows/ \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "product-to-model",
    "steps": [
      {
        "name": "generate-base",
        "type": "generation",
        "config": {
          "prompt": "model wearing product",
          "num_outputs": 4
        }
      },
      {
        "name": "upscale",
        "type": "processing",
        "config": {
          "operation": "upscale",
          "scale": 2
        },
        "depends_on": ["generate-base"]
      },
      {
        "name": "background-removal",
        "type": "processing",
        "config": {
          "operation": "remove-background"
        },
        "depends_on": ["upscale"]
      }
    ]
  }'

# Start workflow
curl -X POST https://compute-api.sartiq.com/api/v1/workflows/{workflow_id}/start \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "image_url": "https://storage.sartiq.com/products/dress.jpg"
    }
  }'

Monitor Workflow Progress

curl https://compute-api.sartiq.com/api/v1/workflows/{workflow_id} \
  -H "Authorization: Bearer $API_KEY"

{
  "id": "wf_abc123",
  "name": "product-to-model",
  "status": "running",
  "current_step": 1,
  "steps": [
    {
      "name": "generate-base",
      "status": "completed",
      "output": {"images": [...]}
    },
    {
      "name": "upscale",
      "status": "running",
      "progress": 0.5
    },
    {
      "name": "background-removal",
      "status": "pending"
    }
  ]
}

---

Monitoring

Check Queue Status

curl https://compute-api.sartiq.com/api/v1/monitoring/queues \
  -H "Authorization: Bearer $API_KEY"

{
  "queues": [
    {
      "name": "generation",
      "pending": 12,
      "running": 8,
      "workers": 10
    },
    {
      "name": "training",
      "pending": 2,
      "running": 2,
      "workers": 4
    }
  ]
}

Check Worker Health

curl https://compute-api.sartiq.com/api/v1/monitoring/workers/health \
  -H "Authorization: Bearer $API_KEY"

{
  "healthy": 14,
  "unhealthy": 1,
  "total": 15,
  "workers": [
    {
      "id": "worker-01",
      "status": "busy",
      "gpu_utilization": 0.85,
      "last_heartbeat": "2024-01-15T10:30:00Z"
    }
  ]
}

---

Error Handling

Handle Task Failures

Python

response = httpx.get(
    f"{base_url}/tasks/{task_id}",
    headers=headers
)
task = response.json()

if task["status"] == "failed":
    error = task.get("error", "Unknown error")

    if "CUDA out of memory" in error:
        # Retry with smaller batch or different worker
        print("GPU memory issue - retrying...")
    elif "Invalid input" in error:
        # Fix input and resubmit
        print(f"Input error: {error}")
    else:
        # Log and alert
        print(f"Task failed: {error}")

Cancel a Task

curl -X POST https://compute-api.sartiq.com/api/v1/tasks/task_abc123/cancel \
  -H "Authorization: Bearer $API_KEY"

{
  "id": "task_abc123",
  "status": "cancelled",
  "cancelled_at": "2024-01-15T10:35:00Z"
}

---

Complete Example: Generate and Download

Python

import httpx
import time
from pathlib import Path

BASE_URL = "https://compute-api.sartiq.com/api/v1"

def generate_images(
    api_key: str,
    image_url: str,
    prompt: str,
    output_dir: str = "./outputs"
) -> list[str]:
    """Generate images and download results."""
    headers = {"Authorization": f"Bearer {api_key}"}

    # 1. Submit task
    response = httpx.post(
        f"{BASE_URL}/tasks/generation",
        headers=headers,
        json={
            "input_image_url": image_url,
            "prompt": prompt,
            "num_outputs": 4
        }
    )
    task_id = response.json()["id"]
    print(f"Task submitted: {task_id}")

    # 2. Wait for completion
    while True:
        response = httpx.get(
            f"{BASE_URL}/tasks/{task_id}",
            headers=headers
        )
        task = response.json()

        if task["status"] == "completed":
            break
        elif task["status"] == "failed":
            raise Exception(task.get("error"))

        print(f"Progress: {task.get('progress', 0) * 100:.0f}%")
        time.sleep(2)

    # 3. Download images
    output_path = Path(output_dir)
    output_path.mkdir(exist_ok=True)

    downloaded = []
    for i, image in enumerate(task["output"]["images"]):
        img_response = httpx.get(image["url"])
        file_path = output_path / f"output_{i}.jpg"
        file_path.write_bytes(img_response.content)
        downloaded.append(str(file_path))
        print(f"Downloaded: {file_path}")

    return downloaded

# Usage
images = generate_images(
    api_key="your-api-key",
    image_url="https://storage.sartiq.com/products/dress.jpg",
    prompt="fashion model, studio lighting, white background"
)