oleander Python SDK

Use the oleander API from Python: run routed lake queries, launch Spark jobs and Spark SQL, run Polars over lake tables, and manage environment variables.

Install

pip install oleanderhq-sdk

Get your API key

Create an API key in oleander 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

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.

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:

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:

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.

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.

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).

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.

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:

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).

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

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

Get Spark cluster information

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.

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.

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.

# 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.

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

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.