wherobots-python-sdk 0.2.1

Python SDK for Wherobots (currently covers the Jobs REST API)

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

Variable	Description	Default
WHEROBOTS_API_KEY	API key	(required)
WHEROBOTS_REGION	Region override (else org default)	(none — org default)
WHEROBOTS_API_BASE_URL	API base URL	https://api.cloud.wherobots.com
WHEROBOTS_VERSION	Wherobots version	latest
WHEROBOTS_REQUEST_TIMEOUT_SECONDS	HTTP request timeout (seconds)	30

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",
)

Parameter	Type	Default
api_key	str | None	None (read from env)
region	str | None	None
base_url	str	"https://api.cloud.wherobots.com"
version	str	"latest"
request_timeout_seconds	int	30

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
    base_url: str | None = None,             # Override API URL
    request_timeout_seconds: int | None = None,  # HTTP timeout
    config: WherobotsConfig | None = None,   # Full config object
)

Instance Methods

Method	Returns	Description
submit()	str	Submit the job. Returns the run ID. Idempotent (returns existing ID if already submitted).
get_status()	RunView	Get current job status and full details.
get_logs(cursor=0, size=100)	LogsResponse	Fetch a page of log entries.
get_metrics()	RunMetricsResponse	Fetch CPU/memory metrics for the run.
get_cpu_utilization()	UtilizationStats	Aggregated CPU utilization (latest, max, avg, series).
get_mem_utilization()	UtilizationStats	Aggregated memory utilization (latest, max, avg, series).
get_cost()	float | None	Total run cost in USD, or None if not yet billed.
get_consumed_spatial_units()	float | None	Spatial Units (SUs) consumed by the run.
refresh()	RunView	Re-fetch from the API and update status/name/runtime/region/version in place.
iter_logs(cursor=0, size=100)	Iterator[dict]	Iterate over all log entries, handling pagination automatically.
poll_for_logs(follow=True, interval=2.0, log_handler=None, max_errors=10)	None	Poll and print logs. If follow=True, continues until job completes. max_errors sets the max consecutive transient errors before giving up.
cancel()	bool	Request cancellation. Returns True on success.
wait_for_completion(poll_interval=5.0, stream_logs=True, log_interval=2.0, max_wait_seconds=None)	JobStatus	Block until the job reaches a terminal state. Set max_wait_seconds to limit wait time (raises WherobotsTimeoutError).
close()	None	Close the underlying HTTP session and release resources.

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

Method	Returns	Description
from_run_id(run_id, api_key=None, ...)	WherobotsJob	Attach to an existing run for read-only log/metric access. No script required. submit() is disabled on the returned instance.
list_runs(...)	RunListPage	List runs with optional filters. No instance required.
add_pypi_dependency(name, version)	dict	Create a PyPI dependency dict for the dependencies parameter.
add_file_dependency(file_path)	dict	Create a file dependency dict (.jar, .whl, .zip, .json).

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",
    },
)

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

Value	Terminal	Description
PENDING	No	Queued
RUNNING	No	Executing
COMPLETED	Yes	Finished OK
FAILED	Yes	Errored out
CANCELLED	Yes	User cancelled

from wherobots import JobStatus

status = JobStatus.RUNNING
status.is_terminal  # False

Runtime

Available compute sizes (API string values):

Enum	Value
Runtime.MICRO	micro
Runtime.TINY	tiny
Runtime.SMALL	small
Runtime.MEDIUM	medium
Runtime.LARGE	large
Runtime.X_LARGE	x-large
Runtime.DOUBLE_X_LARGE	2x-large
Runtime.QUAD_X_LARGE	4x-large
Runtime.MEDIUM_HIMEM	medium-himem
Runtime.LARGE_HIMEM	large-himem
Runtime.X_LARGE_HIMEM	x-large-himem
Runtime.DOUBLE_X_LARGE_HIMEM	2x-large-himem
Runtime.QUAD_X_LARGE_HIMEM	4x-large-himem
Runtime.X_LARGE_HICPU	x-large-hicpu
Runtime.DOUBLE_X_LARGE_HICPU	2x-large-hicpu
Runtime.X_LARGE_MATCHER	x-large-matcher
Runtime.DOUBLE_X_LARGE_MATCHER	2x-large-matcher
Runtime.MICRO_A10_GPU	micro-a10-gpu
Runtime.TINY_A10_GPU	tiny-a10-gpu
Runtime.SMALL_A10_GPU	small-a10-gpu
Runtime.MEDIUM_A10_GPU	medium-a10-gpu
Runtime.LARGE_A10_GPU	large-a10-gpu
Runtime.X_LARGE_A10_GPU	x-large-a10-gpu

from wherobots import Runtime

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

Region

Enum	Value
Region.US_WEST_2	aws-us-west-2
Region.US_EAST_1	aws-us-east-1
Region.EU_WEST_1	aws-eu-west-1
Region.AP_SOUTH_1	aws-ap-south-1

DependencyType

Enum	Value
DependencyType.PYPI	PYPI
DependencyType.FILE	FILE

DependencyFileType

Enum	Value
DependencyFileType.JAR	JAR
DependencyFileType.PYTHON_WHEEL	PYTHON_WHEEL
DependencyFileType.ZIP	ZIP
DependencyFileType.OTHER	OTHER

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.

Field	Type	Description
id	str	Run ID
name	str	Job name
status	JobStatus | None	Current status
runtime	str | None	Compute runtime size
region	str | None	AWS region
version	str | None	Wherobots version
timeout_seconds	int | None	Job timeout
create_time	str | None	ISO-8601 creation timestamp
update_time	str | None	ISO-8601 last update timestamp
start_time	str | None	ISO-8601 start timestamp
complete_time	str | None	ISO-8601 completion timestamp
run_python	RunPythonPayload | None	Python script payload
run_jar	RunJarPayload | None	JAR payload
environment	RunEnvironment | None	Spark config & dependencies
triggered_by	dict | None	User who triggered the run
kube_app	RunKubeApp | None	Kubernetes app details
app_meta	RunAppMeta | None	Spark/Sedona UI URLs
organization	OrganizationCustomer | None	Org info
extra	dict	Any additional API fields

LogsResponse

Returned by get_logs().

Field	Type	Description
items	list[LogItem]	Log entries
next_page	int | str | None	Next page cursor
current_page	int | str | None	Current page cursor

LogItem

Field	Type	Description
raw	str	Raw log line text
timestamp	int | str | None	Epoch int or ISO-8601 string
level	str | None	Log level (INFO, ERROR, etc.)
message	str | None	Parsed message
extra	dict	Additional fields

RunListPage

Returned by list_runs().

Field	Type	Description
items	list[RunView]	Runs on this page
total	int	Total matching runs
next_page	str | None	Next page cursor
previous_page	str | None	Previous page cursor
current_page	str | None	Current page cursor
current_page_backwards	str | None	Backward pagination cursor

RunMetricsResponse

Returned by get_metrics().

Field	Type	Description
series_metrics	dict	Time-series metrics
instant_metrics	dict	Point-in-time metrics
extra	dict	Additional fields

Exceptions

All exceptions inherit from WherobotsJobError.

Exception	Description
WherobotsJobError	Base exception for all SDK errors.
WherobotsAPIError	API returned an error response. Has status_code, response, request_id attributes.
WherobotsValidationError	Client-side validation failed (e.g., invalid name).
WherobotsS3Error	S3 upload or access operation failed. Deprecated — presigned uploads raise WherobotsAPIError instead.
WherobotsConfigError	Configuration is missing or invalid.
WherobotsTimeoutError	Job exceeded the specified timeout.

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

Method	Endpoint	Description
POST	/runs?region=<region>	Create a run
GET	/runs/{id}	Get run details
POST	/runs/{id}/cancel	Cancel a run
GET	/runs/{id}/logs	Get run logs
GET	/runs/{id}/metrics	Get run metrics
GET	/runs	List runs

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

Development

Setup

git clone https://github.com/wherobots/wherobots-python-sdk.git
cd wherobots-python-sdk
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
pre-commit install

Quality checks

make lint        # ruff check + ruff format --check
make format      # auto-fix lint issues and reformat
make typecheck   # mypy (strict)
make check       # lint + typecheck + tests + build

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.