Metadata-Version: 2.5
Name: h2o-connector-service
Version: 0.2.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: 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).

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

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

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

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

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

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

```py
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](https://github.com/h2oai/connector-service/issues/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):

```python
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):

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

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

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

## Supported Data Source Types

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

| `data_source_type` | Display Name         | Category | Worker Language |
|--------------------|----------------------|----------|-----------------|
| `postgresql`       | PostgreSQL           | Tabular  | Go              |
| `bigquery`         | Google BigQuery      | Tabular  | Go              |
| `snowflake`        | Snowflake            | Tabular  | Go              |
| `hive`             | Apache Hive          | Tabular  | Java            |
| `delta-lake`       | Delta Lake           | Tabular  | Rust            |
| `s3`               | Amazon S3            | Blob     | Go              |
| `gcs`              | Google Cloud Storage | Blob     | Go              |
| `azure-blob`       | Azure Blob Storage   | Blob     | Go              |
| `minio`            | MinIO                | Blob     | Go              |
| `h2o-drive`        | H2O Drive            | Blob     | Go              |
| `http`             | HTTP (HTTP/HTTPS/SFTP/FTP) | Blob | Go          |
