Quickstart¶

Install with pip install superb-ai, then authenticate and import a dataset end-to-end.

Auth (3 lines)¶

from superb_ai import Client
client = Client(tenant="acme", api_key="sbd_pk_...")   # or set SUPERB_AI_TENANT + SUPERB_AI_API_KEY
# with both env vars set, `Client()` alone works — nothing to hardcode
assert client.projects.list().items is not None         # smoke: you're in

Machine keys (sbd_pk_…) are the primary credential. The client.auth.* surface is JWT-bootstrap-only (browser login dance) — a key holder never calls it. Each key acts as ONE user; multi-actor workflows need one client per user key.

Client map (namespaces)¶

client. → tenants users api_keys auth account invitations datasets assets projects classes members guidelines labeling_time tags workflow workflow_batch project_assets annotations revisions comments versions exports jobs training models deployments auto_label foundation segment activity visualization

account (my workspaces / accept invitations) is JWT-session-only — an sbd_pk_… key gets 401 there; machine keys already know their workspace. invitations (admin: invite email + role, list, revoke) works with keys.

Pagination: list methods return Page[T] — iterate the page object to auto-fetch all pages (for x in client.annotations.list(pid): ...). include_total=True adds .total. Non-list collection reads (classes, tags, progress, queue stats) return complete models, not pages.

Async jobs: kickoff methods return a JobHandle — call .wait(timeout=...); it raises JobFailedError on failure and JobTimeoutError (job keeps running) on timeout. handle.result_summary carries per-kind outcome keys.

Batch first — the anti-flood rule¶

Never loop a single-item call when N > ~5. Every hot path has a batch twin (sync, explicit ids, ≤1000) and most have a bulk twin (async job, id-list or filter-scoped). Pick by N:

doing this per-item?	≤1000 → sync batch	any size → async bulk (JobHandle)
create annotations	`annotations.asset_batch_create` (1 asset) / `annotations.project_batch_create`	`annotations.import_ndjson`
edit/delete annotations	`annotations.batch_edit` (atomic mixed) / `annotations.batch_delete` (≤500 ids)	`project_assets.bulk_delete_annotations` (+class/source predicates)
assign/approve/reject/unassign	`workflow_batch.batch_*`	`workflow_batch.bulk_*` (filter-scoped)
add/remove project assets	`project_assets.batch_add` / `batch_remove`	`project_assets.bulk_add` / `bulk_remove`
tag assets	`tags.batch_update`	`tags.bulk_update`
delete assets	`assets.batch_delete`	`assets.bulk_delete`
upload files	`assets.upload_paths` (does presign+PUT batching for you)	—
predict on many assets	—	`auto_label.create_run` (never a predict loop)

Flood etiquette (the rate buckets in the Limits section are the budget): - Polling: use the built-in waiters (JobHandle.wait, versions.wait_ready, training.wait_run — they poll at ~3s). Don't hand-roll tighter loops; for long jobs (training, big imports) pass a larger poll_interval. - Counting: include_total=True on one list call — never "iterate and count", and never per-item GETs to check existence (list with a filter instead). - Reading many items: iterate the Page (sequential, one request per page at the max limit) — don't fan out parallel GET-by-id. - Upload concurrency ≤16 on shared/dev tenants; the write bucket is shared by everyone on the tenant.

Worked example: import a COCO dataset end-to-end¶

The full path: upload images → create project with class vocabulary → convert COCO GT → bulk-import → verify. Every step below is load-bearing.

1a. Upload images (never hand-roll the presign dance)¶

from pathlib import Path
ds = client.datasets.create(name="COCO val2017")
paths = sorted(Path("/data/val2017").glob("*.jpg"))
client.assets.upload_paths(ds.id, paths, concurrency=8)   # presign→PUT, batched
# Asset rows are created ASYNCHRONOUSLY by the platform's storage events.
# Poll until they're all ready:
import time
while True:
    ready = client.datasets.stats(ds.id).by_status.get("ready", 0)
    if ready >= len(paths): break
    time.sleep(10)
asset_by_name = {a.filename: a.id for page in [client.assets.list(ds.id, limit=100)]
                 for a in page}  # iterate the Page for ALL assets

Keep concurrency<=16 on shared/dev environments (rate limits are per-tenant).

1b. Create the project + class vocabulary¶

One geometry type per class (allowed_types). 80 COCO classes = 80 class specs:

classes = [{"name": c["name"], "color": f"#{hash(c['name']) & 0xffffff:06x}",
            "allowed_types": ["bbox"]} for c in coco["categories"]]
project = client.projects.create(name="Detection", dataset_id=ds.id,
                                 scope={"all": True}, classes=classes)
class_id = {c.name: c.id for c in client.classes.list(project.id).classes}

Keypoint classes additionally need a topology (COCO person = 17 nodes; edges are COCO's 1-indexed skeleton converted to 0-indexed pairs):

KP_NAMES = ["nose","left_eye","right_eye","left_ear","right_ear","left_shoulder",
  "right_shoulder","left_elbow","right_elbow","left_wrist","right_wrist",
  "left_hip","right_hip","left_knee","right_knee","left_ankle","right_ankle"]
topology = {"version": 1, "nodes": [{"id": i, "name": n} for i, n in enumerate(KP_NAMES)],
            "edges": [[a-1, b-1] for a, b in coco_skeleton_1_indexed]}
classes = [{"name": "person", "color": "#ff0000", "allowed_types": ["keypoint"],
            "keypoint_topology": topology}]

1c. Convert COCO annotations → platform geometry (ALL THREE KINDS)¶

def bbox_geometry(coco_ann):                      # COCO bbox = [x, y, w, h] pixels
    x, y, w, h = coco_ann["bbox"]
    if w <= 0 or h <= 0: return None              # server rejects degenerate boxes
    return {"type": "bbox", "x": x, "y": y, "w": w, "h": h}

def polygon_geometry(coco_ann):
    if coco_ann.get("iscrowd"): return None       # RLE crowds are NOT convertible — SKIP
    parts = []
    for flat in coco_ann["segmentation"]:         # list of flat [x1,y1,x2,y2,...] rings
        pts = list(zip(flat[0::2], flat[1::2]))
        if len(pts) >= 3:                          # rings need >= 3 vertices
            parts.append({"exterior": pts, "holes": []})
    return {"type": "polygon", "polygons": parts} if parts else None

def keypoint_geometry(coco_ann):                  # [x,y,v] * 17; v in {0,1,2} maps 1:1
    if coco_ann.get("num_keypoints", 0) == 0: return None
    k = coco_ann["keypoints"]
    return {"type": "keypoint",
            "points": [{"x": k[i], "y": k[i+1], "visibility": k[i+2]}
                       for i in range(0, len(k), 3)]}

1d. Bulk import (>10k rows → the ASYNC staged path, one call does it all)¶

rows = [{"id": str(uuid.uuid5(NS, str(ann["id"]))),   # deterministic ids => re-import
         "asset_id": asset_by_name[img_name[ann["image_id"]]],  #   is an idempotent upsert
         "class_id": class_id[cat_name[ann["category_id"]]],
         "type": "bbox", "geometry": bbox_geometry(ann)}
        for ann in coco["annotations"] if bbox_geometry(ann)]
job = client.annotations.import_ndjson(project.id, rows, source="imported")
job.wait(timeout=900)
total = client.annotations.list(project.id, limit=1, include_total=True).total
assert total == len(rows)

(≤1000 rows? client.annotations.project_batch_create(project.id, annotations=rows, source="manual") is the sync path — instant, all-or-nothing.)