sandai-operator-client 0.4.4

Client library for the Sandai Operator SDK

Sandai Operator Client

Lightweight Python client for submitting tasks to a Sandai operator over Celery.

Installation

pip install sandai-operator-client

For local development:

pip install -e ".[dev]"

Quick Start

from sandai.operator_client import create_client

client = create_client("video-clipper", "v1")

artifacts, results = client.sync(
    input_artifacts=[{"uri": "file:///tmp/input.mp4", "slot": "video"}],
    options={"clip_duration": 10},
)

Core API

• create_client(operator_name, operator_version, **kwargs) constructs an OperatorClient.
• async_call(...) submits a task and returns a Celery AsyncResult.
• sync(...) submits a task and waits for (artifacts, results).
• desc fetches the operator description payload, validates its status, and caches it per client instance.
• batch_async(...) preserves the legacy best-effort behavior and returns AsyncResult | None per task.
• batch_sync(...) preserves the legacy best-effort behavior and returns legacy tuples per task.
• batch_submit(...) returns structured submission objects with explicit send errors.
• batch_collect(...) resolves structured submission objects into per-task structured results.
• batch_sync_detailed(...) returns structured per-task execution results directly.
• probe() performs a stronger transport/protocol health probe.
• ping() is a compatibility helper that only confirms the submit path is reachable.
• get_queue_info() returns the queue names used for each priority.

Task Controls

The client supports the same task-control fields used by the operator runtime:

• timeout: converted into an absolute deadline before sending the task.
• cancel_key: propagated to the operator so tasks can be skipped before compute starts.
• persistent_output: stored under meta["persistent-output"] for runtimes that preserve uploaded outputs.

Batch APIs

The package now exposes two batch styles:

• Legacy compatibility APIs: batch_async(...) and batch_sync(...) keep the previous best-effort semantics.
• Structured APIs: batch_submit(...), batch_collect(...), and batch_sync_detailed(...) expose explicit per-task submission/execution status instead of collapsing failures into empty results.

Prefer the structured APIs for new integrations.

Health Checks

ping() and probe() are intentionally different:

• ping() returns True when task submission succeeds.
• probe() returns a structured HealthCheckResult with submit_ok, retrieval_ok, response_ok, and optional error details.

Use probe() when you need a meaningful health signal.

Priority Model

Supported priorities are:

• high
• normal
• low
• very_low

Queue names follow this pattern:

<priority>.<operator_name>.<operator_version>

Environment Variables

The client reads these settings when explicit connection arguments are not provided:

• SANDAI_OPERATOR_CELERY_BROKER_URL
• SANDAI_OPERATOR_CELERY_RESULT_BACKEND
• SANDAI_OPERATOR_CELERY_TASK_SERIALIZER
• SANDAI_OPERATOR_CELERY_RESULT_SERIALIZER
• SANDAI_OPERATOR_AUTO_FORGET_RESULT

If broker/backend are not configured, the client falls back to local Redis defaults:

redis://localhost:6379/0

By default the client now submits and stores Celery payloads with msgpack, while advertising accept_content=["json", "msgpack"] so it can still interoperate with older JSON-based operators during rollout.

Error Model

sync() and desc normalize failures into typed exceptions where possible:

• TaskTimeoutError
• TaskExecutionError
• TaskDeadlineExceededError
• TaskCancelledError
• InvalidTaskStatusError
• InvalidResultFormatError
• InvalidPriorityError

Structured batch APIs surface task errors directly on each result object instead of converting them into empty legacy tuples.

Testing

Run the client regression suite from the repository root:

python operator-client/tests/run_all_tests.py

The suite is mock-based and does not require a live Celery or Redis deployment.