oleander Python SDK

Use the oleander API from Python: run 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.

Query (lake)

Run a SQL query against the oleander lake. The second parameter is optional; save defaults to False. Use save=True to persist results as a table.

from oleander_sdk import Oleander, QueryOptions

oleander = Oleander()
result = await oleander.query(
    "SELECT * FROM oleander.default.flowers LIMIT 10",
)
print(result.results.columns, result.results.rows)
print(result.row_count, result.execution_time)
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

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 and RunNotFoundError for structured error handling.