Skip to main content

Connector service Python client

The Connector service Python client (h2o-connector-service) lets you stream data from any configured data source into your Python environment using a few lines of code.

Prerequisites​

Installation​

Install the Python client using pip:

pip install h2o-connector-service

The client already depends on h2o-cloud-discovery and h2o-authn, so you do not need to install them separately.

For streaming to pandas DataFrames or Parquet files, install the matching extras:

pip install "h2o-connector-service[pandas,parquet]"

Authentication​

The client authenticates using your H2O AI Cloud platform token. It reads two environment variables:

# The root URL of your H2O AI Cloud environment (not the Connector Service URL)
export H2O_CLOUD_ENVIRONMENT="https://<your-h2o-cloud-domain>"

# Your platform token
export H2O_CLOUD_CLIENT_PLATFORM_TOKEN="<your-platform-token>"

The client uses the root URL to discover the Connector Service address and to refresh your access token.

Note

In an H2O AI Cloud notebook, these variables are already set. Outside a notebook, you can retrieve your platform token from the H2O AI Cloud interface. See Using a Platform Token for details.

Connecting to the service​

When the two environment variables are set, create a client with no arguments:

import os
from h2o_connector_service import Client

client = Client()

You can also pass the values directly:

client = Client(
    h2o_cloud_url="https://<your-h2o-cloud-domain>",
    refresh_token="<your-platform-token>",
)

Environments on a custom domain​

The client runs discovery only when the environment URL ends in .h2o.ai or .h2o-cloud.com. For any other domain, the client treats the URL as the Connector Service address and sends the token unchanged. In that case, run discovery yourself and pass a token provider:

import os
import h2o_authn
import h2o_discovery
from h2o_connector_service import Client

discovery = h2o_discovery.discover(environment=os.environ["H2O_CLOUD_ENVIRONMENT"])
token_provider = h2o_authn.TokenProvider(
    refresh_token=os.environ["H2O_CLOUD_CLIENT_PLATFORM_TOKEN"],
    issuer_url=discovery.environment.issuer_url,
    client_id=discovery.clients["platform"].oauth2_client_id,
)

client = Client(
    connector_service_url=discovery.services["connector-service"].uri,
    token_provider=token_provider,
)

Streaming tabular data​

To stream data from a tabular source (PostgreSQL, Snowflake, BigQuery, and others), open a session with your connector and worker references, and specify an extraction query:

connector_ref = "connectors/<connector-id>"   # provided by your Workspace Administrator
worker_ref    = "workers/<worker-id>"         # provided by your Workspace Administrator
workspace_id  = "<your-workspace-id>"

with client.open_session(
    workspace_id=workspace_id,
    connector=connector_ref,
    worker=worker_ref,
    extraction={"query": "SELECT id, name, value FROM my_table WHERE active = true"},
) as session:
    session.stream_to_csv("output.csv")

Choose one output per session:

MethodOutput
session.stream_to_csv(path)CSV file
session.stream_to_pandas()pandas DataFrame (needs the pandas extra)
session.stream_to_parquet(path)Parquet file (needs the parquet extra)
session.stream_records()Iterator of records
Note

Each stream_to_* call runs the extraction again. When the stream finishes, the connection completes. To write the same data in two formats, open a second session, or stream once to pandas and convert the DataFrame.

Tip

The connector_ref and worker_ref are resource names in the format connectors/<id> and workers/<id>. Your Workspace Administrator provides these after provisioning the connector and worker for your data source.

Streaming blob (file) data​

To stream files from a blob storage source (S3, MinIO, GCS, Azure, SFTP, and others), use stream_to_files instead. The client reconstructs each file verbatim in a local output directory:

with client.open_session(
    workspace_id=workspace_id,
    connector=connector_ref,
    worker=worker_ref,
    extraction={"paths": {"paths": [{"pattern": "exports/2026/"}]}},
) as session:
    n_files, total_bytes = session.stream_to_files("downloaded_files/")
    print(f"Downloaded {n_files} file(s), {total_bytes} bytes total")

Each file is written under the output directory using its full object key. For example, exports/2026/report.csv is written to downloaded_files/exports/2026/report.csv.

Note

Blob workers stream raw bytes rather than tabular records, so stream_to_csv, stream_to_pandas, and stream_to_parquet do not apply to blob connections. Use stream_to_files for blob data.

Connection lifecycle​

Opening a session with client.open_session(...) creates a Connection, waits for the worker to become ready, and automatically deletes the Connection when the with block exits — even if an error occurs. You do not need to manage the Connection lifecycle manually.

By default, open_session waits up to 300 seconds for the worker. Use wait_timeout to change this.

While a Connection is active, users with permission to list connections in the workspace can see it.

Extraction options​

The extraction parameter controls what data to read from the source. Use exactly one of query, tables, or paths:

OptionDescription
queryA SQL query string (for tabular sources)
tablesOne table specification with optional column and filter constraints (for tabular sources). Only one table is allowed per connection.
pathsA list of object keys or prefixes (for blob sources). A pattern that ends in / or * is a prefix. Any other pattern is an exact object key.
batch_sizeNumber of records per batch (default depends on the data source; usually 1000)
max_unitsMaximum total records or files to read (0 = unlimited)
timeoutTime limit for the whole extraction, for example "600s" (default: no limit)

Example using table extraction with column projection and filtering:

extraction = {
    "tables": {
        "tables": [
            {
                "schema": "public",
                "table": "orders",
                "columns": ["customer", "amount", "status"],
                "filter": "status = 'completed'",
            }
        ]
    },
    "batch_size": 500,
}

Example selecting blob objects. This reads one exact key and every object under one prefix:

extraction = {
    "paths": {
        "paths": [
            {"pattern": "exports/2026/q1/report.csv"},
            {"pattern": "exports/2026/q2/"},
        ]
    },
}

Error handling​

The client raises ConnectorServiceError for API and transport errors. It also raises ConnectorServiceError when the connection fails while the worker is starting, for example because of wrong credentials. The error message includes the cause reported by the platform. When open_session raises, it deletes the Connection before it returns.

from h2o_connector_service import ConnectorServiceError

try:
    with client.open_session(...) as session:
        session.stream_to_csv("output.csv")
except ConnectorServiceError as exc:
    print(f"Connection failed: {exc}")

Next steps​

  • Data Sources — configuration reference for PostgreSQL, Snowflake, S3, and other connectors
  • Secure Credentials — how credentials are securely managed for connectors
Feedback