sqi-sdk (Python client API)

sqi-sdk is the pure-Python library for driving a sqi-server instance from scripts and pipeline tooling — the same operations the web UI performs, over the REST API (with an optional WebSocket extra for live events). Per the roadmap (../ROADMAP.md) it is the foundation for the Phase 2 DCC submitters and for pipeline automation.

This is the reference document. For the project overview and install, see clients/python/README.md; the authoritative wire contract is the OpenAPI spec described in api.md.

Import name: sqi_client · distribution: sqi-sdk
Python 3.9+ · sync-only · core dependency httpx only
Optional extras: yaml (PyYAML), ws (websockets)

Related: sqi-submitter. Building a scripted pipeline tool, a batch submission script, or any headless automation? Use sqi-sdk directly, as below. Building an in-application submitter for Maya, Houdini, Nuke, or Blender? See docs/dcc-submitters.md — the sqi-submitter package depends on sqi-sdk and adds the Qt/native-UI form, scene-context extraction, and pre-fill on top of it.

Installation
Client construction and configuration
Error handling
Pagination: Page vs iter_*
Submitting jobs
Querying and managing jobs
Tasks and logs
Workers
Farm, queue, and resource CRUD
Products
Live events (WebSocket, ws extra)
Conveniences

Installation

pip install sqi-sdk            # core (httpx only)
pip install 'sqi-sdk[yaml]'    # + PyYAML
pip install 'sqi-sdk[ws]'      # + websockets (live event streaming)

For offline installs or a specific pre-release build, the wheel is also attached to each GitHub release — see clients/python/README.md.

Client construction and configuration

SqiClient is constructed from the server's root URL (no /api/v1 suffix — the client adds it). It owns an httpx.Client connection pool and works as a context manager or with an explicit close().

from sqi_client import SqiClient

with SqiClient("http://localhost:8080") as sqi:
    print(sqi.ping())   # True if /healthz is OK
    print(sqi.ready())  # True if /readyz is OK (SQLite + NATS reachable)

Constructor options:

Argument	Default	Purpose
`base_url`	—	Server root, e.g. `http://localhost:8080`. A subpath is preserved (reverse-proxy friendly).
`timeout`	`30.0`	Per-request timeout in seconds.
`token`	`None`	API key or session token, sent as `Authorization: Bearer` on every request and on the WebSocket upgrade. Falls back to `$SQI_TOKEN`, then `$SQI_API_KEY`.
`headers`	`None`	Extra default headers merged into every request. Merged after `token`, so an explicit `Authorization` header wins — useful behind a gateway with its own scheme.
`max_attempts`	`3`	Total attempts for a retried (idempotent GET) request; `1` disables retries.
`retry_backoff`	`0.5`	Base seconds for exponential backoff between retries.
`retry_backoff_max`	`30.0`	Cap on the computed backoff delay.
`retry_jitter`	`True`	Randomize each backoff delay in `[0, computed]`.

Only idempotent GET requests are retried (on connection errors, 5xx, and 429 — honoring Retry-After). POST/PATCH/PUT/DELETE are never retried. Each request logs at DEBUG and each retry at WARNING on the sqi_client logger (no handler is attached — standard library etiquette).

sqi = SqiClient(
    "https://farm.studio.example",
    timeout=15.0,
    token="sqi_…",  # or leave unset and export SQI_API_KEY
    max_attempts=5,
)

When the server has auth enabled, a missing or rejected credential raises SqiAuthError (401/403). Issue an API key from the web UI or POST /api/v1/api-keys — see docs/auth.md.

Error handling

Every error derives from SqiError, so a single except can catch any client failure:

SqiError
├── SqiConnectionError      # server unreachable (DNS, refused, reset)
├── SqiTimeoutError         # request timed out / wait_for_job deadline
└── APIError                # any non-2xx response
    ├── BadRequestError     # 400
    ├── SqiAuthError        # 401 / 403 (missing or rejected credential)
    ├── NotFoundError       # 404
    ├── ConflictError       # 409
    ├── ValidationError     # 422
    ├── RateLimitError      # 429 (see .retry_after)
    └── ServerError         # 5xx

APIError carries status, title, detail, and request_id (parsed from the RFC 7807 application/problem+json body or the X-Request-ID header). str(exc) includes the status, detail, and request ID so failures are diagnosable in logs.

from sqi_client import NotFoundError, ValidationError

try:
    job = sqi.get_job("does-not-exist")
except NotFoundError as exc:
    print(exc.status, exc.detail, exc.request_id)

try:
    sqi.submit_job(bad_template, farm_id=f, queue_id=q)
except ValidationError as exc:
    print(exc.detail)  # e.g. "step 'Render' references undefined parameter 'Frames'"

Pagination: `Page` vs `iter_*`

List endpoints come in two flavors:

list_*(...) returns a single Page[T] mirroring the server's {items, total, limit, offset} wrapper.
iter_*(...) returns a lazy iterator that walks every page for you, advancing offset until the result set is exhausted.

page = sqi.list_jobs(status="running", limit=50)
print(page.total, len(page.items))

for job in sqi.iter_jobs(status="failed"):   # fetches pages on demand
    print(job.id, job.name)

Status filters accept either the enum or its wire string: sqi.list_jobs(status=JobStatus.RUNNING) and sqi.list_jobs(status="running") are equivalent. None filters are omitted from the request entirely.

Note: farms, storage locations, compute locations, usage pools, and products are returned by the server as bare arrays (no pagination), so their list_* methods return a plain list[T]. Only queues, jobs, tasks, and workers are paginated.

Submitting jobs

submit_job(template, *, farm_id, queue_id, owner=None, priority=None, project=None, max_attempts=None, retry_delay_seconds=None, failure_limit=None, depends_on=None) -> Job

The template may be a str (sent verbatim), a pathlib.Path (read from disk), or a dict (serialized to JSON). The Content-Type is chosen automatically — application/json for dicts and JSON-parseable strings/files, application/x-yaml otherwise (and always for .yaml/.yml paths). Serializing a dict needs only the standard library; the yaml extra is not required to submit.

depends_on is an optional list of upstream job IDs (same farm) this job must wait for; if any is not yet completed the returned job's status is blocked instead of pending, and it is released automatically once every upstream completes (or canceled if one fails, is canceled, or is deleted). submit_product_job accepts the same depends_on keyword.

max_attempts, retry_delay_seconds, and failure_limit are optional per-job retry-policy overrides; each left None inherits the queue → farm → server default. They are also accepted by submit_and_wait and submit_product_job, and (as create/update fields) by create_farm/create_queue. The resolved policy is returned on get_job as Job.effective_retry (an EffectiveRetryPolicy with max_attempts/retry_delay_seconds/failure_limit), alongside Job.failed_attempts and, once a job is auto-parked, Job.park_reason.

from pathlib import Path

# From a file on disk:
job = sqi.submit_job(Path("render.yaml"), farm_id=farm_id, queue_id=queue_id)

# From a dict built in code:
job = sqi.submit_job(
    {"specificationVersion": "jobtemplate-2023-09", "name": "My Job", "steps": [...]},
    farm_id=farm_id, queue_id=queue_id, owner="alice", priority=90,
)

A 422 from the server (OpenJD validation failure) is raised as ValidationError with the server's detail preserved verbatim, so a submitter can show it to the artist.

Querying and managing jobs

Method	Description
`list_jobs(*, status, farm_id, queue_id, owner, project, sort_by, sort_dir, limit, offset) -> Page[Job]`	One page of jobs.
`iter_jobs(...) -> Iterator[Job]`	Same filters, auto-paged.
`get_job(job_id) -> Job`	Detailed job with `steps` and `task_counts`; 404 → `NotFoundError`.
`pause_job(job_id) -> Job` / `resume_job(job_id) -> Job`	Pause/resume; returns the updated job.
`set_job_priority(job_id, priority) -> Job`	Validates `priority >= 1` client-side before sending.
`cancel_job(job_id) -> None`	POST `/jobs/{id}/cancel`; returns `None` on `204`.
`retry_job(job_id) -> RetryJobResult`	POST `/jobs/{id}/retry`; revives all `failed`/`canceled` tasks; returns `RetryJobResult(job_id, retried)`.
`delete_job(job_id) -> None`	DELETE; permanently deletes the job and all data (cancels active tasks first); returns `None` on `204`.

job = sqi.get_job(job_id)
print(job.status, job.task_counts.succeeded, "/", job.task_counts.total)

sqi.pause_job(job_id)
sqi.set_job_priority(job_id, 75)
sqi.resume_job(job_id)
sqi.cancel_job(job_id)   # soft-cancel; job data is retained
result = sqi.retry_job(job_id)   # revive all failed/canceled tasks; result.retried = count
sqi.delete_job(job_id)   # hard-delete; all data permanently removed

Tasks and logs

Method	Description
`list_job_tasks(job_id, *, status, sort_by, sort_dir, limit, offset) -> Page[Task]`	One page of a job's tasks.
`iter_job_tasks(job_id, ...) -> Iterator[Task]`	Auto-paged companion.
`get_task(task_id) -> Task`	A single task.
`list_task_attempts(task_id) -> list[TaskAttempt]`	List a task's execution attempts, oldest first — each attempt's status, exit code, and failure message.
`retry_task(task_id) -> RetryResult`	Retry a failed/canceled task; revives the task, its step, and the job (when terminal); response `status` is `ready` or `pending` depending on step dependencies; invalid state → `ConflictError`.
`cancel_task(task_id) -> CancelResult`	Cancel a non-terminal task; terminal task → `ConflictError`.
`get_task_logs(task_id, limit=100, after_nats_seq=0) -> LogPage`	One page of log chunks plus the cursor.
`tail_task_logs(task_id, poll_interval=1.0, from_seq=0, follow=True) -> Iterator[LogChunk]`	Polling log tail.

get_task_logs returns a LogPage whose after_nats_seq is the cursor to pass on the next call to fetch only newer chunks:

cursor = 0
while True:
    page = sqi.get_task_logs(task_id, after_nats_seq=cursor)
    for chunk in page.items:
        print(chunk.data, end="")
    if not page.items:
        break
    cursor = page.after_nats_seq

tail_task_logs does this for you, yielding each chunk in order and sleeping poll_interval between empty polls. With follow=True (the default) it stops once the task reaches a terminal state and the final chunks have been drained; with follow=False it stops at the current end of the log.

for chunk in sqi.tail_task_logs(task_id, follow=True):
    print(chunk.data, end="")

Workers

Method	Description
`list_workers(*, farm_id, queue_id, compute_location, status, sort_by, sort_dir, limit, offset) -> Page[Worker]`	One page of workers.
`iter_workers(...) -> Iterator[Worker]`	Auto-paged companion.
`get_worker(worker_id) -> Worker`	Worker detail, including `current_tasks`.
`disable_worker(worker_id) -> WorkerAction \| None`	Drain and stop new assignments.
`enable_worker(worker_id) -> WorkerAction \| None`	Re-enable a disabled worker.
`remove_worker(worker_id) -> None`	Hard-delete a worker record; `204` → `None`. Only offline workers, or disabled workers whose last heartbeat is older than the heartbeat-timeout window, are removable — an online or live-disabled worker raises `ConflictError`.

for worker in sqi.iter_workers(status="online"):
    print(worker.hostname, worker.status)

action = sqi.disable_worker(worker_id)   # WorkerAction(id=..., status=DISABLED)
sqi.enable_worker(worker_id)

disable_worker/enable_worker return a lightweight WorkerAction (id, status) — the server's response shape — rather than a full Worker.

Farm, queue, and resource CRUD

Each resource family exposes the standard create/list/get/update/delete set. update_* is a full replace (PUT semantics): a field left at its default is reset, so pass every field you want to keep.

Family	Create	List	Get / Update / Delete
Farms	`create_farm(*, name, description=None, max_concurrent_tasks=0, max_attempts=None, retry_delay_seconds=None, failure_limit=None)`	`list_farms() -> list[Farm]`, `iter_farms()`	`get_farm`, `update_farm`, `delete_farm`
Queues	`create_queue(*, farm_id, name, description=None, priority=0, max_concurrent_tasks=0, paused=False, max_attempts=None, retry_delay_seconds=None, failure_limit=None)`	`list_queues(*, farm_id, paused, sort_by, sort_dir, limit, offset) -> Page[Queue]`, `iter_queues(...)`	`get_queue`, `update_queue`, `delete_queue`
Storage locations	`create_storage_location(*, name, description=None, roots=None)`	`list_storage_locations() -> list[StorageLocation]`, `iter_storage_locations()`	`get_storage_location`, `update_storage_location`, `delete_storage_location`
Compute locations	`create_compute_location(*, name, description=None)`	`list_compute_locations() -> list[ComputeLocation]`, `iter_compute_locations()`	`get_compute_location`, `update_compute_location`, `delete_compute_location`
Usage pools	`create_usage_pool(*, name, max_concurrent, server_hint=None)`	`list_usage_pools() -> list[UsagePool]`, `iter_usage_pools()`	`get_usage_pool`, `update_usage_pool`, `delete_usage_pool`

farm = sqi.create_farm(name="studio-a", description="Studio A render farm")
queue = sqi.create_queue(farm_id=farm.id, name="renders", priority=50, max_concurrent_tasks=200)
loc = sqi.create_storage_location(
    name="project-root", roots={"on-prem": "/mnt/farm"}
)  # type is derived by the server from the roots
pool = sqi.create_usage_pool(name="arnold-pool", max_concurrent=10)

create_usage_pool/update_usage_pool validate max_concurrent >= 1 client-side (raising ValueError) before sending. Every UsagePool response also carries read-only, server-computed in_use (active claims) and available (max(max_concurrent - in_use, 0)) fields for live utilization.

Products

Products are named, versioned wrappers around OpenJD templates stored in the server's catalog. Eight methods cover them:

Method	Description
`list_products() -> list[Product]`	Every product (built-ins + custom); bare array, no pagination.
`iter_products() -> Iterator[Product]`	Iterator companion.
`get_product(name) -> Product`	One product by name; 404 → `NotFoundError`.
`create_product(*, name, template, format, title=None, description=None, category=None, version=None) -> Product`	Create a custom product from a raw OpenJD template.
`update_product(name, *, template, format, title=None, description=None, category=None, version=None) -> Product`	Full PUT replacement of a custom product.
`delete_product(name) -> None`	Delete a custom product; a built-in raises `SqiAuthError` (403).
`get_product_parameters(name) -> list[ProductParameter]`	The parsed job parameters — type, default, allowed values, and `user_interface` hints. 404 → `NotFoundError`; an unparseable stored template → `ValidationError` (422).
`submit_product_job(name, *, farm_id, queue_id, job_name=None, owner=None, submitter=None, priority=None, project=None, parameters=None, max_attempts=None, retry_delay_seconds=None, failure_limit=None, depends_on=None) -> Job`	Submit a job from a product.

products = sqi.list_products()
params = sqi.get_product_parameters("python")
for p in params:
    print(p.name, p.type, p.default)

job = sqi.submit_product_job(
    "python",
    farm_id=farm_id, queue_id=queue_id,
    job_name="My Script Run",
    parameters={"Script": "print('hello')", "Interpreter": "python3"},
)

submit_product_job uses the keyword job_name= (not name=) so it does not shadow the positional product name; the wire field sent to the server is "name".

Live events (WebSocket, `ws` extra)

With pip install 'sqi-sdk[ws]', SqiClient.events() opens a live event stream over GET /api/v1/ws. The stream is a context manager; subscribe to one or more subjects and iterate Event objects. It tracks the last seq per subject and auto-reconnects + resubscribes (resuming from that seq) across an idle disconnect or server restart.

with sqi.events() as stream:
    stream.subscribe("workers")
    stream.subscribe(f"jobs/{job_id}/tasks")
    for event in stream:
        print(event.subject, event.seq, event.payload)

Subjects: jobs, jobs/{job-id}/tasks, tasks/{task-id}/logs, workers, and diagnostics (see api.md for payload shapes). Subscribing to diagnostics requires the diagnostics.read permission. A failed subscription or server error frame is raised as SqiError.

tail_task_logs_live(task_id, from_seq=0) -> Iterator[LogChunk] is the WebSocket-backed counterpart to tail_task_logs:

for chunk in sqi.tail_task_logs_live(task_id):
    print(chunk.data, end="")

Without the ws extra installed, events() itself succeeds — the websockets import is deferred to connect(), so the ImportError naming the exact remedy (pip install 'sqi-sdk[ws]') is raised when the stream is entered as a context manager, and on the first iteration of tail_task_logs_live. The live tasks/{id}/logs payload omits id/nats_seq/received_at, so those are zero-valued on the yielded LogChunk; seq_num and the content fields are set.

Conveniences

wait_for_job(job_id, poll_interval=2.0, timeout=None) -> Job polls get_job until the job reaches a terminal status (completed, failed, canceled), returning the final Job. It raises SqiTimeoutError if timeout elapses first. Note that paused is not terminal — pass a timeout if a job might be paused indefinitely.

submit_and_wait(template, *, farm_id, queue_id, owner=None, priority=None, project=None, max_attempts=None, retry_delay_seconds=None, failure_limit=None, depends_on=None, poll_interval=2.0, timeout=None) -> Job composes submit_job + wait_for_job for the simplest pipeline script (a depends_on-blocked job simply polls through blocked → pending → running → terminal like any other):

from pathlib import Path

job = sqi.submit_and_wait(
    Path("render.yaml"), farm_id=farm_id, queue_id=queue_id, timeout=3600
)
if job.status != JobStatus.COMPLETED:   # compare by value, not identity
    raise SystemExit(f"job {job.id} ended as {job.status}")

me() -> Principal returns the authenticated principal — subject, display_name, roles, permissions, kind, and (for local accounts) username — from GET /api/v1/auth/me. Gate on permissions, not roles: check for "jobs.submit_as" before setting a job owner other than your own user; without it the server responds 403. With auth disabled the anonymous superuser principal is returned.

server_version() -> ServerVersion returns the server's build metadata from GET /api/v1/version.

Navigation Menu

sqi-sdk (Python client API)

Contents

Installation

Client construction and configuration

Error handling

Pagination: `Page` vs `iter_*`

Submitting jobs

Querying and managing jobs

Tasks and logs

Workers

Farm, queue, and resource CRUD

Products

Live events (WebSocket, `ws` extra)

Conveniences

FilesExpand file tree

python-client.md

Latest commit

History

python-client.md

File metadata and controls

sqi-sdk (Python client API)

Contents

Installation

Client construction and configuration

Error handling

Pagination: Page vs iter_*

Submitting jobs

Querying and managing jobs

Tasks and logs

Workers

Farm, queue, and resource CRUD

Products

Live events (WebSocket, ws extra)

Conveniences

Pagination: `Page` vs `iter_*`

Live events (WebSocket, `ws` extra)