h2o-connector-service

Python client for the H2O Connector Service. Use it to create connectors, open connections, and stream data to and from a data source.

pip install h2o-connector-service

Connect

H2O AI Cloud

Pass your cloud URL and refresh token. The client finds the connector service URL and refreshes the access token for you.

from h2o_connector_service import Client

client = Client(
    h2o_cloud_url="https://cloud.h2o.ai",
    refresh_token="<your refresh token>",
)

Inside an H2O notebook the env vars H2O_CLOUD_ENVIRONMENT and H2O_CLOUD_CLIENT_PLATFORM_TOKEN are already set, so you need no arguments:

client = Client()

A known service URL

Pass connector_service_url= when you already know the address of the connector service itself. The client uses the URL as given, never runs discovery, and sends the token as a static bearer token.

client = Client(
    connector_service_url="https://connector-service.cloud.h2o.ai",
    refresh_token="<a valid access token>",
)

Add verify_ssl=False for a local or test deployment that uses a self-signed certificate.

Your own token provider

If your service already manages OIDC tokens, pass a callable that takes no arguments. The client calls it on every request, so returning a fresh access token keeps token expiry invisible to the SDK. You cannot use refresh_token= and token_provider= together.

client = Client(
    connector_service_url="https://connector-service.cloud.h2o.ai",
    token_provider=lambda: my_auth_layer.current_access_token(),
)

Service identity

A platform service that runs in the same Kubernetes cluster can authenticate with its ServiceAccount token. The client sends the token in the x-h2o-service-authorization header, and the server checks it against its allowlist. This path needs connector_service_url=.

def read_projected_sa_token() -> str:
    with open("/var/run/secrets/kubernetes.io/serviceaccount/token") as f:
        return f.read().strip()

# Service only. Every operation belongs to the service identity.
client = Client(
    connector_service_url="https://connector-service.cloud.h2o.ai",
    service_token_provider=read_projected_sa_token,
)

# On behalf of a user. The service vouches for the call and the user token
# gives the identity. If the user token is invalid the server rejects the
# request. It never falls back to the service identity.
client = Client(
    connector_service_url="https://connector-service.cloud.h2o.ai",
    service_token_provider=read_projected_sa_token,
    token_provider=lambda: current_user_access_token(),
)

whoami() needs a user credential. It raises ValueError on a service-only client.

Workspace id

Workspace id is not a constructor argument. Pass it on every workspace-scoped call, for example client.connectors.list(workspace_id) and client.open_session(workspace_id=..., ...).

Read data

A connection needs three parts:

• a Connector — the data source config
• a Worker, created from a WorkerTemplate. A dynamic Worker starts one pod per connection.
• an ExtractionConfig — what to read

Connectors and Workers are long-lived. A platform admin creates them once. End users only call open_session(...), which creates one connection per stream. See examples/quickstart.py for a runnable version.

from h2o_connector_service import Client

client = Client(h2o_cloud_url="https://cloud.h2o.ai", refresh_token="...")
workspace = "my-workspace"

# ── ADMIN: create the long-lived resources ───────────────────────────────

# 1. WorkerTemplate (global) — image and pod defaults
wt = client.worker_templates.create(
    metadata={"name": "wt-pg"},
    image="353750902984.dkr.ecr.us-east-1.amazonaws.com/h2oai-connectorservice-workerpostgresql:v0.3.0",
    pull_policy="IfNotPresent",  # K8s short form or the full IMAGE_PULL_POLICY_* enum
    supported_data_source_types=["postgresql"],
    default_resources={"cpu": "250m", "memory": "512Mi"},
    enabled=True,
)

# 2. Connector (workspace) — data source type and driver-native config
connector = client.connectors.create(
    workspace,
    metadata={"name": "pg"},
    data_source_type="postgresql",
    data_source_config={
        "PGHOST": "db.example.com",
        "PGPORT": "5432",
        "PGDATABASE": "mydb",
        "PGUSER": "postgres",
        "PGPASSWORD": "secret",  # read this from env or SecureStore in real code
    },
)

# 3. Worker (workspace) — built from the WorkerTemplate above
worker = client.workers.create(
    workspace,
    metadata={"name": "w-pg"},
    worker_template=f"workerTemplates/{wt.metadata.name}",
)

# ── END USER: open one session, then stream ──────────────────────────────

# 4. open_session creates the Connection, waits for WORKER_READY, and on exit
#    deletes only the Connection. The other resources stay.
with client.open_session(
    workspace_id=workspace,
    connector=f"connectors/{connector.metadata.name}",
    worker=f"workers/{worker.metadata.name}",
    extraction={"query": "SELECT * FROM my_table", "batch_size": 100},
) as session:
    for row in session.stream_records():
        print(row)

# 5. Delete the long-lived resources when you no longer need them.
worker.delete()
connector.delete()
wt.delete()

To read from blob storage, pass paths instead of a query:

extraction = {"paths": {"paths": [{"pattern": "data/"}]}}

A pattern ending in / or * is a prefix (only a trailing * works, there are no globs). Any other pattern is one exact object key.

Output formats

A session can write the stream straight into a file or a frame. Pick one output per session. Each call runs the extraction again, and a dynamic worker's connection completes after one read. The calls below are alternatives:

# CSV file. Rows are written as they arrive, so memory stays flat.
session.stream_to_csv("output.csv")

# pandas DataFrame. Needs: pip install h2o-connector-service[pandas]
df = session.stream_to_pandas()

# Parquet file, written in row-group chunks.
# Needs: pip install h2o-connector-service[parquet]
session.stream_to_parquet("output.parquet")

# datatable Frame, built with chunked rbind.
# Needs: pip install h2o-connector-service[datatable]
frame = session.stream_to_data_table()

# H2O Frame. Needs a running H2O cluster and h2o.init().
# Needs: pip install h2o-connector-service[h2o]
h2o_frame = session.stream_to_h2o_frame()

# Blob files, written to a directory.
file_count, byte_count = session.stream_to_files("out_dir")

Write data

open_write_session creates every resource it needs, waits for the worker, and deletes them all on exit. It also needs a worker image, either from worker_image= as below or from an env var. See Worker image.

with client.open_write_session(
    "postgresql",
    pg_cfg,
    workspace_id="my-workspace",
    target_table="my_schema.my_table",
    worker_image="353750902984.dkr.ecr.us-east-1.amazonaws.com/h2oai-connectorservice-workerpostgresql:v0.3.0",
) as session:
    session.write_records(
        records,
        target_table="my_schema.my_table",
        mode="UPSERT",
        conflict_columns=["id"],
    )

Write modes are INSERT, UPSERT, APPEND, and REPLACE. The default is INSERT, and UPSERT also needs conflict_columns.

Pass mode, conflict_columns, batch_size, and schema to write_records, not to open_write_session. The session is the long-lived handle and each write carries its own settings. open_write_session accepts the same argument names and checks that mode is valid, but it does not apply them, so setting mode only there still writes with INSERT.

Worker image

open_write_session and open_blob_write_session need a container image to start the worker pod. The SDK has no built-in default image, on purpose. An image name without a registry prefix makes kubelet pull from docker.io/library/<name>, which fails on every managed cluster with ErrImagePull and no useful error message.

The client picks the image in this order:

1. The worker_image= argument, for a single call.
2. The H2O_CONNECTOR_SERVICE_DEFAULT_WORKER_IMAGE_<TYPE> env var, set once at deployment time. It must include the registry.

If neither is set, the call fails at once with a ConnectorServiceError that names the env var to set. The <TYPE> suffix is the connector_type you passed, in upper case, with hyphens replaced by underscores. So "postgresql" reads ..._POSTGRESQL and "delta-lake" reads ..._DELTA_LAKE.

Set the env var once per deployment, in the pod spec, in Helm values, or in the shell:

export H2O_CONNECTOR_SERVICE_DEFAULT_WORKER_IMAGE_POSTGRESQL=\
353750902984.dkr.ecr.us-east-1.amazonaws.com/h2oai-connectorservice-workerpostgresql:v0.3.0

Note the 353750902984.dkr.ecr.us-east-1.amazonaws.com/ prefix. Always write the registry and the organization in full, even for a Docker Hub image (docker.io/<org>/...). A bare h2oai-connectorservice-workerpostgresql:v0.3.0 is what kubelet reads as docker.io/library/..., which is the failure this section warns about.

Use worker_image= when different call sites need different images, for example when you test a new worker build. It overrides the env var.

Optional dependencies

Install extras for the output formats you need:

pip install h2o-connector-service[pandas]       # pandas DataFrames
pip install h2o-connector-service[parquet]      # Parquet files (pyarrow)
pip install h2o-connector-service[datatable]    # datatable Frames
pip install h2o-connector-service[h2o]          # H2O Frames (pandas + pyarrow + h2o)