h2o-connector-service

• Python client: https://pypi.org/project/h2o-connector-service/
• Source: https://github.com/h2oai/connector-service

Python client SDK for the H2O Connector Service. Provides a high-level API to create connectors, open connections, and stream extracted data from supported data sources (PostgreSQL, Snowflake, Hive, Delta Lake, Blob Storage, and more).

pip install h2o-connector-service

Quick Start (H2O Cloud Discovery)

When h2o_cloud_url= looks like an H2O AI Cloud URL (*.h2o.ai / *.h2o-cloud.com), the Client runs OIDC discovery and exchanges your refresh token automatically:

from h2o_connector_service import Client

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

The same construction is also available with no args when the env vars H2O_CLOUD_ENVIRONMENT and H2O_CLOUD_CLIENT_PLATFORM_TOKEN are set:

client = Client()

Workspace id is NOT a constructor parameter — pass it per call on every workspace-scoped operation (client.connectors.list(workspace_id), client.open_session(workspace_id=..., ...)).

Quick Start (Direct Connector Service URL)

When you already know the connector-service URL (local dev, Kind, in-cluster service DNS, or any deployment that does not use H2O Cloud discovery), pass connector_service_url= — it is used verbatim, discovery is never run regardless of URL shape, and the token is used as a static bearer:

client = Client(
    connector_service_url="https://connector-service.h2oai.test",
    refresh_token="<a valid access token>",
    verify_ssl=False,
)

(Legacy behavior: a non-h2o-cloud-shaped h2o_cloud_url= also short-circuits discovery, but that relies on URL-shape sniffing — hosts under *.h2o.ai / *.h2o-cloud.com would trigger discovery. Prefer connector_service_url= when you mean a direct URL.)

Custom Token Provider (Embedding in Backend Services)

Backend services that manage their own OIDC token lifecycle (per-user tokens, external refresh machinery) can pass a zero-arg callable instead of a refresh token. The callable is invoked on every request, so returning a freshly refreshed access token makes expiry transparent to the SDK. refresh_token= and token_provider= are mutually exclusive; with token_provider=, h2o_authn is never imported:

client = Client(
    connector_service_url="https://connector-service.my-domain",
    token_provider=lambda: my_auth_layer.current_access_token(),
)

token_provider= also composes with h2o_cloud_url= if you want discovery to resolve the service URL while keeping token refresh caller-owned.

Service Identity (Trusted In-Cluster Services)

Platform services running in the same Kubernetes cluster as connector-service can authenticate with their ServiceAccount token instead of (or in addition to) a user token. The token is sent as the x-h2o-service-authorization header on every request; the server authorizes it against its serviceAuth.serviceAccounts allowlist. Requires connector_service_url= (no discovery on this path).

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

# Service-only: all operations attribute to the service identity.
client = Client(
    connector_service_url="https://connector-service.my-domain",
    service_token_provider=read_projected_sa_token,
)

# On-behalf-of: the service identity vouches for the call; the user token
# supplies the identity the server attributes operations to. 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.my-domain",
    service_token_provider=read_projected_sa_token,
    token_provider=lambda: current_user_access_token(),
)

whoami() requires a user credential and raises ValueError on a service-only client.

End-to-End Streaming Flow

A connection has three required pieces — a Connector (data source config), a Worker (a pod backed by a WorkerTemplate), and an ExtractionConfig (what to extract). Connectors + Workers are durable infrastructure provisioned by a platform admin; end users only create per-stream Connections via client.open_session(...). See examples/quickstart.py for a runnable version.

from h2o_connector_service import Client

client = Client(h2o_cloud_url="...", refresh_token="...")
workspace = "my-workspace"

# ── ADMIN: provision durable infrastructure ──────────────────────────────

# 1. WorkerTemplate (global-scoped) — image + pod defaults
wt = client.worker_templates.create(
    metadata={"name": "wt-pg"},
    image="<registry>/h2oai-connectorservice-workerpostgresql:latest",
    pull_policy="IfNotPresent",  # K8s shorthand or full IMAGE_PULL_POLICY_* enum
    supported_data_source_types=["postgresql"],
    default_resources={"cpu": "250m", "memory": "512Mi"},
    enabled=True,
)

# 2. Connector (workspace-scoped) — data_source_type + 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 from env / SecureStore in real code
    },
)

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

# ── END USER: one open_session call, then stream ─────────────────────────

# 4. open_session creates the Connection, waits for WORKER_READY, and on
#    exit deletes ONLY the Connection — the durable infra is untouched.
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. Tear down the durable resources when they are no longer needed.
worker.delete()
connector.delete()
wt.delete()

Worker Image Configuration

open_write_session / open_blob_write_session need a container image name to create the worker pod. There is deliberately no built-in default image table: a bare image name (no registry prefix) makes Kubelet pull from docker.io/library/<name>, which fails on every managed cluster (EKS, GKE, AKS) with ErrImagePull and no actionable diagnostic (issue #539).

The image resolves in this order (first hit wins):

1. worker_image= kwarg — explicit per-call override (Client.open_write_session(…, worker_image=…)).
2. H2O_CONNECTOR_SERVICE_DEFAULT_WORKER_IMAGE_<TYPE> env var — set once at deployment time (registry-qualified); applies to every call for that connector type.

If neither path yields an image, the call fails fast with a ConnectorServiceError naming the H2O_CONNECTOR_SERVICE_DEFAULT_WORKER_IMAGE_<TYPE> env var to set.

The supported <TYPE> suffixes (uppercase) are POSTGRESQL, SNOWFLAKE, BLOB, DELTA, HIVE, and HTTP.

Pattern 1 — per-deployment env-var override (set once, applies to every call):

import os
# Set once, applies to all open_write_session("postgresql", ...) calls in this process.
os.environ["H2O_CONNECTOR_SERVICE_DEFAULT_WORKER_IMAGE_POSTGRESQL"] = (
    "123456789.dkr.ecr.us-east-1.amazonaws.com/h2oai-connectorservice-workerpostgresql:v1.38.0"
)

The env var is also a clean fit for Deployment.spec.template.spec.containers[].env, Helm values.yaml, or any config-management tool — there is no Python-side wiring required.

Pattern 2 — per-call explicit worker_image= (overrides the env var):

with client.open_write_session(
    "postgresql",
    pg_cfg,
    workspace_id="my-workspace",
    target_table="my_schema.my_table",
    worker_image="123456789.dkr.ecr.us-east-1.amazonaws.com/h2oai-connectorservice-workerpostgresql:v1.38.0",
) as session:
    session.write_records(records, target_table="my_schema.my_table")

Use Pattern 2 when different workspaces or call-sites need different images (e.g., A/B testing a new worker build) — it cleanly overrides any process-wide env-var default.

Output Formats

Once you have a session, stream data into various formats:

# CSV file (memory-safe — rows written as they arrive)
session.stream_to_csv("output.csv")

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

# Parquet file (memory-safe, chunked row groups)
# requires: pip install h2o-connector-service[parquet]
session.stream_to_parquet("output.parquet")

# datatable Frame (memory-safe, chunked rbind)
# requires: pip install h2o-connector-service[datatable]
frame = session.stream_to_data_table()

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

Optional Dependencies

Install extras for additional output format support:

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)

Supported Data Source Types

The authoritative list is served by client.data_source_profiles.list(). Currently: