sqi-sdk

Pure-Python client for sqi — a distributed task and render farm manager.

sqi-sdk (import name sqi_client) is a pure-Python library for programmatic job submission, status queries, and management. It covers the same operations as the web UI via the REST API, and is the foundation for the Phase 2 DCC submitters and for pipeline automation scripts (see ../../ROADMAP.md).

It talks to a running sqi-server over its REST API, with an optional WebSocket extra for live event streaming. The only required dependency is httpx; everything else is an opt-in extra, so the library stays light enough to embed in DCC Python environments (Maya, Houdini, Nuke).

Requirements

• Python 3.9 or newer (VFX Reference Platform CY2022+; covers Maya 2023+, Houdini 19.5+, Nuke 14+).
• A reachable sqi-server instance.

Installation

pip install sqi-sdk            # core (httpx only)
pip install 'sqi-sdk[yaml]'    # + PyYAML (for your own YAML handling; not needed to submit)
pip install 'sqi-sdk[ws]'      # + websockets for live event streaming

For offline installs or a specific pre-release build, the wheel is also attached to each GitHub release:

pip install https://github.com/uberware/sqi/releases/download/vX.Y.Z/sqi_sdk-X.Y.Z-py3-none-any.whl
# with an extra:
pip install "sqi-sdk[ws] @ https://github.com/uberware/sqi/releases/download/vX.Y.Z/sqi_sdk-X.Y.Z-py3-none-any.whl"

The package ships a py.typed marker, so type checkers see its annotations.

Quickstart

from pathlib import Path

from sqi_client import SqiClient

with SqiClient("http://localhost:8080") as sqi:
    # Submit an OpenJD template (a Path is read from disk; a str is sent verbatim;
    # a dict is serialized to JSON) and block until the job finishes.
    job = sqi.submit_and_wait(
        Path("render.yaml"), farm_id="<farm-id>", queue_id="<queue-id>", timeout=600
    )
    print("job", job.id, "->", job.status)

    # Print each task's captured log output.
    for task in sqi.iter_job_tasks(job.id):
        page = sqi.get_task_logs(task.id)
        print("".join(chunk.data for chunk in page.items), end="")

What you can do

• Submit raw OpenJD job templates (submit_job, submit_and_wait).
• Query jobs, tasks, workers, and logs with typed models and automatic pagination (list_* returns a Page; iter_* walks every page lazily).
• Manage jobs (pause, resume, set priority, cancel, retry) and tasks (cancel, retry) and workers (enable, disable).
• CRUD farms, queues, storage locations (type is derived from roots by the server), and usage pools.
• Tail logs by polling (tail_task_logs) or live over WebSocket with the ws extra (tail_task_logs_live).

Errors map to a typed hierarchy rooted at SqiError (e.g. NotFoundError, ValidationError, ConflictError, SqiTimeoutError). The transport retries idempotent GETs with backoff.

Authentication

sqi-server can optionally require authentication (auth.enabled, off by default — see docs/auth.md). SqiClient already has bearer-token support wired in for when a server requires it:

from sqi_client import SqiAuthError, SqiClient

# Pass a token explicitly...
sqi = SqiClient("http://localhost:8080", token="<token>")

# ...or let it fall back to $SQI_TOKEN, then $SQI_API_KEY, from the environment.
sqi = SqiClient("http://localhost:8080")  # reads SQI_TOKEN / SQI_API_KEY if set

try:
    sqi.list_products()
except SqiAuthError:
    print("authentication required or credential rejected")

When a token is available (explicit token= or either environment variable), it is sent as Authorization: Bearer <token> on every request; an explicit headers={"Authorization": ...} argument still overrides it. A 401 or 403 response raises SqiAuthError.

Issue a key from the web UI (Admin → API Keys) or POST /api/v1/api-keys; the raw key is shown once at creation. Provide it to the SDK as token= (or via $SQI_TOKEN / $SQI_API_KEY) and it authenticates headlessly against a server with auth.enabled=true. Browser sessions stay cookie-based; API keys are the machine credential. See docs/auth.md for the full model.

Discovering your permissions

client.me() returns the authenticated Principal, including permissions. Check for "jobs.submit_as" before setting a job owner other than your own user — without it the server responds 403.

Products

Products are named, versioned wrappers around OpenJD templates that live in the server's catalog. The SDK exposes eight methods for working with them (iter_products() is the iterator companion to list_products()):

with SqiClient("http://localhost:8080") as sqi:
    # List all products (built-ins + custom), no pagination
    products = sqi.list_products()

    # Fetch one product by name
    product = sqi.get_product("python")

    # Create a custom product from a raw OpenJD template
    custom = sqi.create_product(
        name="my-render",
        title="My Renderer",
        template="specificationVersion: jobtemplate-2023-09\nname: My Renderer\nsteps: []\n",
        format="yaml",
        description="Render a frame range.",
        readme="# My Renderer\n\nLonger usage notes in Markdown go here.\n",
        category="Rendering",
        version="1.0.0",
    )

    # Replace a custom product's fields (full PUT replacement)
    sqi.update_product(
        "my-render",
        template=custom.template,
        format="yaml",
        title="My Renderer v2",
    )

    # Delete a custom product (built-ins return 403 Forbidden)
    sqi.delete_product("my-render")

    # Fetch the parsed job parameters for a product (type, default, UI hints)
    params = sqi.get_product_parameters("python")
    for p in params:
        print(p.name, p.type, p.default)

    # Submit a job from a product; job_name overrides the template's job name
    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"},
        max_attempts=5,  # optional per-job retry overrides; omit to inherit
    )
    print("submitted job", job.id)

A product has two independent text fields: description is a short (max 500 character) plain-text blurb -- not Markdown, since it reaches consumers that cannot render markup, such as the Blender addon's tooltip -- and it is the field product search matches on. readme is long-form Markdown (max 8000 characters), rendered only on the product's detail page in the web UI; it is never searched. Both caps are enforced server-side (HTTP 400 on create/update) and are not validated client-side.

update_product is a full PUT replacement: any field omitted from the call, including readme, is cleared on the server rather than left unchanged.

get_product_parameters raises NotFoundError when the product does not exist and ValidationError when the stored template cannot be parsed (HTTP 422). submit_product_job uses the keyword argument job_name= (not name=) to avoid shadowing the positional product name argument; the wire field sent to the server is "name". submit_product_job also accepts optional owner, submitter, priority, project, and the retry overrides max_attempts, retry_delay_seconds, failure_limit; each is sent only when set and otherwise inherits the queue → farm → server default.

Documentation

Full reference — construction and configuration, every public method with examples, error handling, pagination, log tailing, and the conveniences — is in docs/python-client.md. Runnable examples live in examples/.

License

AGPL-3.0-or-later. See the repository root for the full license text.