Wherobots Python SDK

The official Python SDK for Wherobots. Currently wraps the Wherobots Jobs REST API — submit, monitor, cancel, and inspect Spark/Sedona job runs from Python. Additional Wherobots surfaces will be added over time.

Features

• Submit Python scripts or JARs as Wherobots jobs
• Auto-upload local scripts via presigned URLs (only needs an API key)
• Stream or iterate logs with pagination
• List runs with status, name, and date filters
• Typed dataclass models for all API responses (no pydantic dependency)
• Environment-based configuration

Installation

pip install wherobots-python-sdk

Quickstart

The simplest way to run a local script — only an API key is needed:

from wherobots import WherobotsJob

job = WherobotsJob(
    script="my_analysis.py",  # local file — auto-uploaded via presigned URL
    name="analysis-job-001",
    runtime="tiny",
    region="aws-us-west-2",
    api_key="wbk_...",  # or set WHEROBOTS_API_KEY env var
)

run_id = job.submit()
print("Run ID:", run_id)

# Block until done, streaming logs to stdout
job.wait_for_completion(stream_logs=True)

You can also pass an S3 URI directly if the script is already uploaded:

job = WherobotsJob(
    script="s3://my-bucket/scripts/analysis.py",
    name="analysis-job-001",
    runtime="tiny",
)

Script Upload Behavior

When script is a local file path (not an s3:// URI) and auto_upload=True (the default), the SDK uploads your script via a presigned S3 URL to Wherobots managed storage. Only needs a valid API key — no AWS credentials required.

You can also reference scripts already in a Storage Integration bucket by passing the S3 URI directly:

job = WherobotsJob(
    script="s3://my-integration-bucket/scripts/my_analysis.py",
    name="analysis-job-001",
    runtime="tiny",
    auto_upload=False,
)

Authentication

The SDK uses API key authentication via the X-API-Key header.

job = WherobotsJob(
    script="s3://bucket/script.py",
    name="my-job-001",
    api_key="wbk_...",
)

The API key can also be set via the WHEROBOTS_API_KEY environment variable.

Configuration

Environment Variables

WherobotsConfig

from wherobots.config import WherobotsConfig

# Build from env vars with optional overrides
config = WherobotsConfig.from_env(
    api_key="wbk_...",
    region="aws-us-west-2",
)

# Or construct directly
config = WherobotsConfig(
    api_key="wbk_...",
    region="aws-us-west-2",
    base_url="https://api.cloud.wherobots.com",
    request_timeout_seconds=30,
    version="latest",
)

Client attribution

Every request the SDK sends carries an advisory X-Wherobots-Client header so Wherobots can attribute workloads to the client that started them:

X-Wherobots-Client: client=python-sdk;ver=0.3.0;plat=darwin

The header is an ordered, comma-separated chain of hops — leftmost is the origin client, and each component appends its own hop on the right. If your tool embeds this SDK, pass your own hop as client_chain and the SDK appends itself after it:

job = WherobotsJob(
    script="job.py",
    name="my-job-name",
    client_chain="client=airflow;ver=1.2.0",
)
# X-Wherobots-Client: client=airflow;ver=1.2.0, client=python-sdk;ver=0.3.0;plat=linux

client_chain also works on WherobotsJob.from_run_id(...) and WherobotsConfig, or can be set once via WHEROBOTS_CLIENT_CHAIN. The header is advisory only: it is used for attribution and analytics and never affects authentication or authorization.

SDK Reference

WherobotsJob

The primary class for managing job runs.

Constructor

WherobotsJob(
    script: str,                             # Path or S3 URI to .py or .jar
    name: str,                               # Job name (8-255 chars, [a-zA-Z0-9_\-.]+)
    runtime: str | Runtime | None = None,    # Compute size (None -> org default)
    region: str | Region | None = None,      # Region; str passed as-is (None -> org default)
    api_key: str | None = None,              # API key (or env var)
    version: str | None = None,              # "latest" | "preview"
    timeout_seconds: int = 3600,             # Job timeout
    args: list[str] | None = None,           # Script arguments
    spark_configs: dict[str, str] | None = None,  # Spark config overrides
    dependencies: list[dict] | None = None,  # PyPI or file dependencies
    spark_driver_disk_gb: int | None = None, # Driver disk size (GB)
    spark_executor_disk_gb: int | None = None, # Executor disk size (GB)
    jar_main_class: str | None = None,       # Main class (required for JARs)
    auto_upload: bool = True,                # Upload local files to S3 (managed storage)
    base_url: str | None = None,             # Override API URL
    request_timeout_seconds: int | None = None,  # HTTP timeout
    config: WherobotsConfig | None = None,   # Full config object
    client_chain: str | None = None,         # Upstream X-Wherobots-Client chain
)

Instance Methods

WherobotsJob supports the context manager protocol for automatic cleanup:

with WherobotsJob(script="s3://bucket/script.py", name="my-job") as job:
    run_id = job.submit()
    job.wait_for_completion()
# session is automatically closed

Alias: Job is a convenience alias for WherobotsJob:

from wherobots import Job
job = Job(script="s3://bucket/script.py", name="my-job")

Class Methods

Attaching to an Existing Run

If you already have a run_id (from the CLI, the Wherobots UI, or a prior SDK session) you can attach without a script:

from wherobots import WherobotsJob

job = WherobotsJob.from_run_id("run-abc-123")
print(job.status, job.name)

# Stream remaining logs
job.poll_for_logs(follow=False)

# Or paginate
for entry in job.iter_logs(size=200):
    print(entry["raw"])

# Aggregated utilization for a completed run
cpu = job.get_cpu_utilization()
print(f"CPU peak {cpu.max}, avg {cpu.avg}, samples {len(cpu.series)}")

# Billing
print(f"cost ${job.get_cost():.2f}, SUs {job.get_consumed_spatial_units()}")

Calling job.submit() on an attached instance raises WherobotsValidationError — these instances are read-only.

Listing Runs

from wherobots import WherobotsJob, JobStatus

page = WherobotsJob.list_runs(
    api_key="wbk_...",
    region="aws-us-west-2",
    status=[JobStatus.RUNNING, JobStatus.PENDING],
    name_pattern="etl-*",
    size=20,
)

for run in page.items:
    print(f"{run.id}  {run.name}  {run.status.value}")

# Paginate
if page.next_page:
    next_page = WherobotsJob.list_runs(cursor=page.next_page)

Dependencies

from wherobots import WherobotsJob

job = WherobotsJob(
    script="s3://bucket/script.py",
    name="dep-test-job",
    dependencies=[
        WherobotsJob.add_pypi_dependency("pandas", "2.0.0"),
        WherobotsJob.add_pypi_dependency("numpy", "1.24.0"),
        WherobotsJob.add_file_dependency("s3://bucket/libs/my_lib-1.0-py3-none-any.whl"),
    ],
    spark_configs={
        "spark.executor.memory": "4g",
        "spark.executor.cores": "2",
    },
)

Local Dependency Files

add_file_dependency also accepts a local path to a .whl, .jar, .zip, or .json. The SDK uploads the file just before the job is submitted, fetching short-lived AWS STS credentials from the Wherobots API (POST /storage/credentials) and using boto3 multipart under the hood — no AWS credentials need to leave your environment.

job = WherobotsJob(
    script="my_job.py",  # local script auto-upload still works
    name="job-with-local-wheel",
    dependencies=[
        # Local wheel — uploaded to managed storage at submit() time
        WherobotsJob.add_file_dependency("dist/mypkg-0.1.0-py3-none-any.whl"),
        # Already-uploaded artifact — used as-is
        WherobotsJob.add_file_dependency("s3://bucket/shared-libs/lib-2.0.jar"),
        WherobotsJob.add_pypi_dependency("pandas", "2.0.3"),
    ],
)
job.submit()

Auto-upload always targets the org's managed storage:

Local dependency uploads honor the same auto_upload flag as the script — if auto_upload=False, attempting to submit a job with a local dependency path raises WherobotsValidationError. The 5 GiB cap on a single upload applies; for larger artifacts, pre-upload and pass an s3:// URI.

Uploading to a Storage Integration

WherobotsJob stays focused on job orchestration — it doesn't take any "where files go" kwargs. To upload to a Storage Integration or a custom subdirectory, use FilesAPI directly and pass the resulting s3:// URI to WherobotsJob:

from wherobots import FilesAPI, WherobotsJob
from wherobots.config import WherobotsConfig

with FilesAPI.from_config(WherobotsConfig.from_env()) as files:
    si = files.resolve_integration("customer-bucket")
    # Build explicit destinations; upload_file is the "upload this file
    # to this exact s3 path" primitive.
    script_uri = files.upload_file(
        "my_job.py",
        files.dest_uri_for(si, "my_job.py", "job-scripts/sdk-e2e"),
    )
    wheel_uri = files.upload_file(
        "dist/internal-lib-1.0.whl",
        files.dest_uri_for(si, "internal-lib-1.0.whl", "job-scripts/sdk-e2e"),
    )

job = WherobotsJob(
    script=script_uri,  # already-uploaded URI
    name="job-in-integration",
    dependencies=[WherobotsJob.add_file_dependency(wheel_uri)],
)
job.submit()

Both artifacts land under s3://<integration.path>/job-scripts/sdk-e2e/.

FilesAPI is also where to look for low-level controls (custom managed subdirs, raw s3:// destinations, STS credential fetch for BYO upload tooling). See FilesAPI.upload_file, upload_managed_file, upload_dependency, dest_uri_for, and get_storage_credentials.

JAR Jobs

job = WherobotsJob(
    script="s3://bucket/jars/my-app.jar",
    name="jar-job-001",
    jar_main_class="com.example.Main",
    args=["--input", "s3://data/input", "--output", "s3://data/output"],
    runtime="medium",
)

Enums

JobStatus

from wherobots import JobStatus

status = JobStatus.RUNNING
status.is_terminal  # False

Runtime

Available compute sizes (API string values):

from wherobots import Runtime

job = WherobotsJob(
    script="s3://bucket/script.py",
    name="gpu-job-001",
    runtime=Runtime.SMALL_A10_GPU,
)

Region

DependencyType

DependencyFileType

Models

All models are Python dataclasses with from_dict() and to_dict() methods for JSON round-tripping.

RunView

Returned by get_status() and list_runs(). Represents a single job run.

LogsResponse

Returned by get_logs().

LogItem

RunListPage

Returned by list_runs().

RunMetricsResponse

Returned by get_metrics().

Exceptions

All exceptions inherit from WherobotsJobError.

from wherobots import WherobotsAPIError, WherobotsJobError

try:
    job.submit()
except WherobotsAPIError as e:
    print(f"HTTP {e.status_code}: {e}")
    print(f"Request ID: {e.request_id}")
except WherobotsJobError as e:
    print(f"Error: {e}")

Storage Integrations

The SDK can run scripts stored in Storage Integrations — user-configured S3 buckets registered with your Wherobots organization. Pass the S3 URI directly as the script parameter:

job = WherobotsJob(
    script="s3://my-integration-bucket/scripts/pipeline.py",
    name="pipeline-job-001",
    runtime="small",
    auto_upload=False,
)

To discover your integration paths programmatically:

from wherobots.api.files import FilesAPI
from wherobots.config import WherobotsConfig

config = WherobotsConfig.from_env()
with FilesAPI.from_config(config) as files_api:
    for si in files_api.list_integrations():
        print(f"{si.name}: {si.path} ({si.region})")

To set up a Storage Integration, follow the S3 Storage Integration guide.

API Endpoints

Base URL: https://api.cloud.wherobots.com

Development

Setup

git clone https://github.com/wherobots/wherobots-python-sdk.git
cd wherobots-python-sdk
uv sync --group dev
uv run pre-commit install

Quality checks

make lint        # ruff check + ruff format --check
make format      # auto-fix lint issues and reformat
make typecheck   # pyrefly (strict config)
make rails       # import-linter + no-suppression + ast-grep
make check       # lint + typecheck + rails + tests + build

Suppression baseline

Pre-existing debt is grandfathered in .slop-suppressions.json (CODEOWNED): the inline suppressions already in the tree, plus the # noqa: CODE / # pyrefly: ignore comments that ruff check --add-noqa and pyrefly check --suppress-errors wrote for pre-existing violations. The suppressions are functional at the tool level; the keystone (.slop-repellant/check_no_suppression.py) fails any net-new one.

• Prune after fixing baselined code (drops stale entries):
  • ruff's RUF100 auto-removes stale noqas via the pre-commit hook's --fix
  • uv run pyrefly suppress --remove-unused
  • then uv run python .slop-repellant/check_no_suppression.py --update
• Reset (re-record from scratch): delete .slop-suppressions.json, then run the same --update.
• Shrinking the baseline is always welcome; growing it requires CODEOWNER sign-off — fix the code instead of widening the baseline.

Releasing

See RELEASING.md for the release flow (version bump → GitHub release → automated PyPI publish).

Running Tests

make test               # unit tests (integration tests skipped by default)

# Or directly:
pytest -o "addopts="    # override pyproject.toml --cov flags if pytest-cov missing
pytest -v               # verbose
pytest --cov=wherobots --cov-report=term-missing  # with coverage

Integration Tests

Integration tests run against the live Wherobots API and are skipped by default. To run them:

export WHEROBOTS_API_KEY="your-key"
make test-integration

# Or pass the key via CLI option:
pytest -m integration -v -o "addopts=" --wherobots-api-key="your-key"

Linting & Formatting

make format             # auto-format with ruff
make lint               # ruff check + ruff format --check (same as CI)

Before You Push

make check              # lint + test + build — mirrors the CI pipeline

License

Apache-2.0. See LICENSE for details.