Metadata-Version: 2.5
Name: oleanderhq-sdk
Version: 0.8.0
Summary: Python SDK for oleander: run routed lake queries, introspect catalogs and table schemas, launch Spark jobs and Spark SQL, run Polars, manage environment variables
License-Expression: MIT
Keywords: data,lake,oleander,openLineage,polars,query,sdk,serverless,spark
Requires-Python: >=3.10
Requires-Dist: httpx<1,>=0.27
Requires-Dist: pydantic<3,>=2.7
Provides-Extra: dev
Requires-Dist: pytest-asyncio<1,>=0.24; extra == 'dev'
Requires-Dist: pytest<9,>=8; extra == 'dev'
Requires-Dist: respx<1,>=0.22; extra == 'dev'
Description-Content-Type: text/markdown

# oleander Python SDK

Use the [oleander](https://oleander.dev) API from Python: run routed lake queries, launch Spark jobs and Spark SQL, run Polars over lake tables, and manage environment variables.

## Install

```bash
pip install oleanderhq-sdk
```

## Get your API key

Create an API key in [oleander settings](https://oleander.dev/app/settings), or run `oleander configure` if you use the CLI. You can pass the key when creating the client or set the `OLEANDER_API_KEY` environment variable.

## Quick start

```python
import asyncio
from oleander_sdk import Oleander

async def main():
    oleander = Oleander()
    result = await oleander.list_spark_jobs()
    print(result.artifacts)

asyncio.run(main())
```

## API

All methods are async. Use `await` when calling them.

### Run a query

`query_run` sends the query to the unified query router. Oleander parses the SQL, estimates how much data the referenced tables hold, and picks the engine (`duckdb`, `polars`, `bloom`) and machine size to match, so leave `engine` as `auto` unless you want a specific one. The choice and the reasoning behind it come back in `engine_decision`.

```python
from oleander_sdk import Oleander, QueryRunOptions, QueryTable

oleander = Oleander()
result = await oleander.query_run(
    "SELECT * FROM oleander.default.flowers LIMIT 10",
)
print(result.results.columns, result.results.rows)
print(result.row_count, result.execution_time)
print(result.engine_decision.engine, result.engine_decision.reasons)
```

`query_run` is read-only: SQL that could change data is rejected before the request goes out. Use `query_submit` for those.

Pass `explain=True` to see which engine a query would take, its estimated input size, and whether your plan allows it, without running anything or spending compute:

```python
result = await oleander.query_run(
    QueryRunOptions(sql="SELECT * FROM oleander.default.events", explain=True)
)
print(result.engine_decision.engine, result.engine_decision.size_band)
```

Pass `script` instead of `sql` to run a Polars DataFrame script that assigns `result`, listing the tables it reads in `tables`:

```python
scripted = await oleander.query_run(
    QueryRunOptions(
        script="result = events.group_by('day').len()",
        tables=[QueryTable(alias="events", table="default.events")],
    )
)
```

### Submit a query that writes

`query_submit` covers everything that changes data: a SELECT plus a `destination` to write it to, and a statement that names its own target (INSERT, UPDATE, DELETE, MERGE, DDL) with no destination. It also takes reads too large to return interactively.

Whether the write finishes on the call depends on the engine the router picked, so read `state`: `COMPLETE` means it already landed, `SUBMITTED` means a job is running and `run_id` is there to poll. Result rows are never returned, only `row_count` when it is known.

```python
from oleander_sdk import QuerySubmitAndWaitOptions, QuerySubmitOptions

submitted = await oleander.query_submit(
    QuerySubmitOptions(
        sql="SELECT day, count(*) AS n FROM oleander.default.events GROUP BY day",
        destination="default.daily_counts",
        write_mode="overwrite",  # or "append"
    )
)
print(submitted.state, submitted.output_table, submitted.run_id)

# Or submit and poll until an asynchronous run finishes. A write that lands
# inline returns immediately with `run` unset.
waited = await oleander.query_submit_and_wait(
    QuerySubmitAndWaitOptions(
        sql="INSERT INTO oleander.default.daily_counts SELECT * FROM staging.daily",
    )
)
print(waited.state)  # COMPLETE | FAIL | ABORT
```

### Query (lake, DuckDB only)

`query` predates the router and always runs DuckDB, with optional auto-save by query hash. Prefer `query_run`; use this only for `save=True`.

```python
from oleander_sdk import Oleander, QueryOptions

oleander = Oleander()
result = await oleander.query(
    "SELECT * FROM oleander.default.flowers LIMIT 10",
    QueryOptions(save=True),
)
if result.saved_table_name:
    print("Saved to:", result.saved_table_name)
```

### List Spark jobs

List your Spark artifacts. Options: `limit` (default 20), `offset` (default 0).

```python
from oleander_sdk import Oleander, ListSparkJobsOptions

oleander = Oleander()
result = await oleander.list_spark_jobs()
print(result.artifacts, result.has_more)

next_page = await oleander.list_spark_jobs(ListSparkJobsOptions(offset=20))
```

### Launch Spark job

Submit a Spark job. Required: `namespace`, `name`, `entrypoint`. `cluster` defaults to `"oleander"`. The legacy `script_name` argument is still accepted as an alias for `entrypoint`.

```python
from oleander_sdk import Oleander, SparkJobSubmitOptions

oleander = Oleander()
result = await oleander.submit_spark_job(SparkJobSubmitOptions(
    namespace="my-namespace",
    name="my-job-name",
    entrypoint="my_script.py",
))
print("Run ID:", result.run_id)
```

To target an external cluster, set `cluster` and provide the cluster-specific properties that match the current API:

```python
result = await oleander.submit_spark_job(SparkJobSubmitOptions(
    cluster="emr-prod",
    namespace="my-namespace",
    name="my-job-name",
    entrypoint="s3://bucket/jobs/main.py",
    args=["--date", "2026-03-11"],
    py_files="s3://bucket/deps.zip",
    packages=["org.example:my-lib:1.0.0"],
))
```

### Wait for a run to finish

Submit and poll until the run completes. Optional: `poll_interval_ms` (default 10000), `timeout_ms` (default 600000).

```python
from oleander_sdk import Oleander, SubmitSparkJobAndWaitOptions

oleander = Oleander()
result = await oleander.submit_spark_job_and_wait(SubmitSparkJobAndWaitOptions(
    namespace="my-namespace",
    name="my-job-name",
    entrypoint="my_script.py",
))
print(result.run_id, result.state)  # COMPLETE | FAIL | ABORT
```

### Get run status

```python
run = await oleander.get_run(run_id)
print(run.state, run.duration)
```

### Get Spark cluster information

```python
cluster = await oleander.get_spark_cluster("emr-prod")
print(cluster.type, cluster.properties)
```

### Spark SQL

Submit a Spark SQL query that writes its result to an Iceberg table. Required: `namespace`, `name`, `query`, `output_table`. Optional: `write_mode` (`"OVERWRITE"` default, or `"APPEND"`), `driver_machine_type`, `executor_machine_type`, `executor_numbers`.

```python
from oleander_sdk import Oleander, SparkSqlSubmitOptions, SubmitSparkSqlAndWaitOptions

oleander = Oleander()
submitted = await oleander.submit_spark_sql(SparkSqlSubmitOptions(
    namespace="my-namespace",
    name="nightly-agg",
    query="SELECT day, count(*) AS n FROM oleander.default.events GROUP BY day",
    output_table="default.daily_counts",
))
print(submitted.run_id, submitted.state)  # SUBMITTED

# Or submit and poll until the run finishes:
result = await oleander.submit_spark_sql_and_wait(SubmitSparkSqlAndWaitOptions(
    namespace="my-namespace",
    name="nightly-agg",
    query="SELECT day, count(*) AS n FROM oleander.default.events GROUP BY day",
    output_table="default.daily_counts",
    write_mode="APPEND",
))
print(result.state)  # COMPLETE | FAIL | ABORT
```

### Polars

Run a Polars SQL query (with table bindings) or a Python script over lake tables. Query mode requires at least one table; pass `distributed=True` plus a `destination` table to run on Polars Cloud. Use `destination`/`save_mode` to persist results.

```python
from oleander_sdk import Oleander, PolarsOptions, PolarsTable

oleander = Oleander()

# SQL query mode
result = await oleander.polars(PolarsOptions(
    query="SELECT day, count(*) AS n FROM events GROUP BY day",
    tables=[PolarsTable(alias="events", table="default.events")],
))
print(result.results.columns, result.results.rows)

# Script mode
scripted = await oleander.polars(PolarsOptions(
    script=my_polars_script,  # Python source using polars
    params={"start_date": "2026-07-01"},
    destination="default.polars_out",
    save_mode="overwrite",
))
if scripted.saved:
    print("Wrote", scripted.saved.rows_written, "rows")
```

### Catalog introspection

Discover Iceberg catalogs, namespaces, tables, and table metadata. Catalog defaults to the built-in `oleander` catalog; namespace defaults to `default`.

```python
# Registered catalogs (name, type, properties). Pass include_tables=True to
# also walk every catalog and return catalog/namespace/table triples (slower).
catalogs = await oleander.list_catalogs()

namespaces = await oleander.list_catalog_namespaces()  # or ("my_catalog")

# All tables in a namespace, or across every namespace when omitted
tables = await oleander.list_catalog_tables(namespace="default")

# Current schema as a flat field list (id, name, type, required)
schema = await oleander.get_catalog_table_schema("events")
print([(f.name, f.type) for f in schema.fields])

# Raw Iceberg metadata: location, schemas, partition-specs, snapshots, properties
metadata = await oleander.get_catalog_table_metadata("events")
print(metadata.table["location"])

# Size of a table (or a partition subset); size_bytes/record_count are strings
size = await oleander.get_catalog_table_size(
    "events",
    partition_filters=[{"key": "day", "value": "2026-07-14"}],
)
```

### Environment variables

Manage organization environment variables (available to Spark jobs and scripts). Names are normalized to uppercase; setting an existing name overwrites its value.

```python
await oleander.set_environment_variable("MY_TOKEN", "secret-value")

all_vars = await oleander.list_environment_variables()
one = await oleander.get_environment_variable("MY_TOKEN")

await oleander.delete_environment_variable(name="MY_TOKEN")
# or by id: await oleander.delete_environment_variable(id=one.id)
```

### Typed error handling

The SDK raises structured errors for HTTP failures:

- `OleanderHttpError` for any non-2xx response (`status`, `method`, `path`, `url`, `body`, `api_error`, `api_details`)
- `RunNotFoundError` (subclass of `OleanderHttpError`) when `get_run(run_id)` returns 404
- `QueryBillingError` (subclass of `OleanderHttpError`) when the query router refuses on billing grounds. Retrying the same query fails identically in all three cases:
  - `PlanUpgradeRequiredError` — the query is fine, the plan does not reach the engine or input size (`plan`, `required_plan`)
  - `PaymentMethodRequiredError` — the engine the router picked is metered compute and needs a card on file
  - `CreditLimitExceededError` — the organization is past its credit limit and every query is blocked

```python
from oleander_sdk import Oleander, OleanderHttpError, RunNotFoundError

try:
    await oleander.get_run(run_id)
except RunNotFoundError as err:
    print("Run is not visible yet:", err.run_id)
except OleanderHttpError as err:
    print(err.status, err.path, err.api_error or err.api_details)
```

## Options

- **Constructor**: `Oleander(api_key=..., base_url=...)`. Omit `api_key` to use `OLEANDER_API_KEY`. Set `base_url` to use a different endpoint (e.g. `http://localhost:3000`).
- **Models**: The package exports Pydantic models (e.g. `OleanderOptions`, `SparkJobSubmitOptions`) if you want to validate config or options yourself.
- **Errors**: The package exports `OleanderHttpError`, `RunNotFoundError`, and the `QueryBillingError` family for structured error handling.
