Metadata-Version: 2.5
Name: h2o-connector-service
Version: 0.3.0
Summary: Python client SDK for the H2O Connector Service — create connectors, open connections, and stream extracted data
Project-URL: Source, https://github.com/h2oai/connector-service
Project-URL: Issues, https://github.com/h2oai/connector-service/issues
Author-email: "H2O.ai, Inc." <support@h2o.ai>
Keywords: blob-storage,connector,data-extraction,data-ingestion,delta-lake,gRPC,h2o,hive,postgresql,snowflake
Classifier: Development Status :: 4 - Beta
Classifier: License :: Other/Proprietary License
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: googleapis-common-protos>=1.72
Requires-Dist: grpcio>=1.64
Requires-Dist: h2o-authn>=3.1.0
Requires-Dist: h2o-cloud-discovery>=3.3.0
Requires-Dist: httpx>=0.27
Requires-Dist: protobuf>=4.25
Requires-Dist: pydantic>=2
Provides-Extra: datatable
Requires-Dist: datatable>=1.0; extra == 'datatable'
Provides-Extra: dev
Requires-Dist: grpcio-tools>=1.64; extra == 'dev'
Requires-Dist: mypy-protobuf>=3.6; extra == 'dev'
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=1.0; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=9.0.3; extra == 'dev'
Requires-Dist: respx>=0.21; extra == 'dev'
Requires-Dist: ruff==0.15.22; extra == 'dev'
Requires-Dist: types-grpcio>=1.0; extra == 'dev'
Provides-Extra: h2o
Requires-Dist: h2o>=3.44; extra == 'h2o'
Requires-Dist: pandas>=1.5; extra == 'h2o'
Requires-Dist: pyarrow>=14; extra == 'h2o'
Provides-Extra: pandas
Requires-Dist: pandas>=1.5; extra == 'pandas'
Provides-Extra: parquet
Requires-Dist: pyarrow>=14; extra == 'parquet'
Provides-Extra: test
Requires-Dist: pytest-asyncio>=1.0; extra == 'test'
Requires-Dist: pytest-cov>=5.0; extra == 'test'
Requires-Dist: pytest>=9.0.3; extra == 'test'
Requires-Dist: respx>=0.21; extra == 'test'
Description-Content-Type: text/markdown

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

```bash
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.

```py
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:

```py
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.

```py
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.

```py
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=`.

```py
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** — a pod created from a **WorkerTemplate**
- 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.

```py
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="docker.io/h2oai/h2oai-connectorservice-workerpostgresql:v1.38.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:

```py
extraction = {"paths": {"paths": [{"pattern": "data/*.parquet"}]}}
```

## Output formats

A session can write the stream straight into a file or a frame:

```py
# 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](#worker-image).

```py
with client.open_write_session(
    "postgresql",
    pg_cfg,
    workspace_id="my-workspace",
    target_table="my_schema.my_table",
    worker_image="docker.io/h2oai/h2oai-connectorservice-workerpostgresql:v1.38.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:

```bash
export H2O_CONNECTOR_SERVICE_DEFAULT_WORKER_IMAGE_POSTGRESQL=\
docker.io/h2oai/h2oai-connectorservice-workerpostgresql:v1.38.0
```

Note the `docker.io/h2oai/` prefix. Even for a Docker Hub image you must write the registry and the organization in
full. A bare `h2oai-connectorservice-workerpostgresql:v1.38.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:

```bash
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)
```
