Python SDK - CorpHub User

The CorpHub User SDK is an asynchronous Python library for consuming the connectors and services registered in the Corporate Hub. It allows you to read and write enterprise data, and to invoke Corporate Hub services, directly from your own code.

The SDK is intended for applications that consume the Corporate Hub, for example dashboards, LLM agents, Spark jobs, and notebooks. With it, you can:

• Query relational connectors in read-only mode, and stream large result sets page by page.
• Write to writable connectors, either with DML statements or with batch upserts.
• Search, upsert, and delete vectors in the vector indexes exposed by a vector connector.
• Send natural-language queries to an Agent service registered in the Corporate Hub.
• Call tools, resources, and prompts exposed by an external MCP server registered in the Corporate Hub.
• Request field mappings from the semantic mapper against a loaded ontology.
• Provision a new database with the provisioner, and register it as a writable connector.
• Register, publish, and retrieve versioned ontologies from the ontology catalog of your Organization.

A single CorphubClient instance is the entry point. It resolves a typed handle for each service and communicates with the platform over REST through the Gateway, so you do not manage the internal transport yourself.

Note that this SDK is for consuming the Corporate Hub. To register a new component with the Corporate Hub instead, such as a connector, a mapper, or an agent service, use the Component SDK.

Requires Python 3.12 or later.

Install

The SDK is published as corphub-user-sdk:

pip install "corphub-user-sdk"          # core SDK
pip install "corphub-user-sdk[spark]"   # + PySpark integration
pip install "corphub-user-sdk[all]"     # everything

From a local checkout (PDM):

cd user-sdk-python
pdm install -d

Transport

All data operations are sent over REST to the Corporate Hub Gateway. Each call is submitted with POST /api/v1/requests and then polled with GET /api/v1/requests/{requestId} until it completes.

The Gateway handles the internal communication with the orchestrator on your behalf. As a result, the SDK holds no message broker connection and applies no body encryption. The connection between the SDK and the Gateway relies on transport security (TLS).

The registry, security gate, and data catalog REST APIs are reached through the same Gateway, using its reverse proxy under /registry, /sec-gate, and /data-catalog. The SDK therefore never needs more than one base URL.

To set the Gateway base URL, use one of the following:

• The gateway_url= parameter of CorphubClient.
• The CORPHUB_GATEWAY_URL environment variable (default http://localhost:8600).

Quick start

import asyncio
from user_sdk import CorphubClient, VectorQueryBuilder, FilterRelation


async def main() -> None:
    async with CorphubClient(
        gateway_url="http://localhost:8600",   # or CORPHUB_GATEWAY_URL
        auth_token="...",          # GAM/bearer token
        role="analyst",
    ) as client:
        # Relational (read-only) connector
        ds = await client.relational_connector("my-relational-config-id")
        graphs = await ds.get_schema()
        async for row in ds.execute_query("SELECT * FROM users LIMIT 10"):
            print(row)

        # Vector connector
        vec = await client.vector_connector("my-vector-config-id")
        for index in await vec.get_schema():       # list[VectorIndex]
            print(index.name, index.dimension)
        query = (
            VectorQueryBuilder()
            .embedding([0.1, 0.2, 0.3])
            .top_k(5)
            .filter("genre", FilterRelation.EQ, "drama")
            .build()
        )
        matches = await vec.search(query, index_name="movies")

        # Agent
        agent = await client.agent("my-agent-config-id")
        answer = await agent.query("How many users signed up last week?", source="my-relational-config-id")


asyncio.run(main())

CorphubClient

CorphubClient is the entry point. It is an async context manager — use async with so the gateway HTTP client is released on exit; otherwise call await client.close() yourself. Closing the client also cancels the in-flight background prefetches of any open RowIterator its data sources produced:

client = CorphubClient(auth_token="...", role="analyst", gateway_url="...")
try:
    ds = await client.relational_connector("cfg")
    ...
finally:
    await client.close()

The publisher (gateway HTTP client) is created lazily on the first connector call and reused for the lifetime of the client.

Constructor

CorphubClient(
    auth_token: str,
    role: str,
    registry_token: str | None = None,
    gateway_url: str | None = None,
)

CorphubClient does not currently expose an active-tenant selector: its constructor takes no tenant argument, so it builds the low-level REST clients without one and requests resolve to the caller's primary tenant. To target a non-primary tenant, construct the REST clients standalone with tenant= — see Tenant selection.

Factory methods

Each factory below is async, looks up config_id in the registry (with pagination handled and results cached per client instance), and verifies the backing service is the expected type before returning a handle — so a mismatched config_id fails immediately instead of surfacing as a confusing request timeout later.

CorphubClient also exposes four low-level REST clients directly, already wired to the gateway URL and auth token — see Low-level REST clients:

client.registry      # RegistryApi
client.sec_gate      # SecGateApi
client.data_catalog      # DataCatalogApi
client.audit      # AuditApi

Relational connectors (read-only)

RelationalDataSource (from client.relational_connector(...)) exposes:

async def get_schema() -> list[Graph]
def execute_query(sql: str, page_size: int = 500, prefetch_depth: int = 2) -> RowIterator
def execute_entity_scan(entity_name: str, page_size: int = 500, prefetch_depth: int = 2) -> RowIterator

• get_schema() returns Graph objects (name, entities: list[Entity], relations: list[Relation]); each Entity has name and fields: list[EntityField] (name, data_type, is_nullable, is_primary_key, is_foreign_key, and max_length: int | None — the declared column length for sized types like varchar(255), or None).
• execute_query runs SQL against the connector; execute_entity_scan reads a named entity directly (no SQL) for connectors whose native extraction path doesn't go through a query engine.
• Both return a RowIterator immediately — no request is sent until you start iterating.

RowIterator — paginated streaming with prefetch

rows = ds.execute_query("SELECT * FROM orders", page_size=200, prefetch_depth=2)
async for row in rows:            # row: dict[str, Any]
    ...
print(rows.total_count)           # populated after the first page is fetched
await rows.close()                # cancel any in-flight background prefetches

• Fetches pages of page_size rows and prefetches up to prefetch_depth pages ahead in the background while you consume the current one — bounded by an internal buffer, so memory stays proportional to prefetch_depth.
• total_count is None until the first page returns; it's set once and trusted only from that first response.
• Also usable as an async context manager (async with ds.execute_query(...) as rows:), which calls close() for you on exit.
• If a background prefetch fails, the exception surfaces from the next anext() call (i.e. the next loop iteration), not immediately.
• Prefer async with or an explicit await rows.close() to release an iterator deterministically. Closing the owning CorphubClient also cancels the in-flight prefetches of every iterator its data sources handed out, and a GC'd iterator with pending prefetches cancels them best-effort while emitting a ResourceWarning.

Writable connectors

WritableDataSource (from client.writable_connector(...)) extends RelationalDataSource — you get get_schema() / execute_query() / execute_entity_scan() for free, plus:

async def execute_update(sql: str) -> WriteResult
async def write_batch(entity: str, rows: list[ dict]) -> WriteResult

• execute_update runs a DML statement (INSERT/UPDATE/DELETE).
• write_batch upserts a batch of dict rows into entity.
• Both return WriteResult(rows_affected: int, status: str, error: str | None).

result = await ds.write_batch("orders", [{"customer_id": 101, "amount": 299.99}])
print(result.rows_affected, result.status)

Vector connectors

VectorDataSource (from client.vector_connector(...)) is writable only when the backing config is a WritableConnector; otherwise upsert/delete raise ReadOnlyConnectorError. Check vec.is_read_only if you need to branch on it.

async def get_schema() -> list[VectorIndex]
async def search(query: VectorQuery, index_name: str | None = None) -> list[VectorMatch]
async def upsert(vectors: list[dict], index_name: str | None = None) -> dict
async def delete(ids: list[str] | None = None, filter_expr: dict | None = None, index_name: str | None = None) -> dict

index_name is required for search/upsert/delete (raises ValueError if omitted).

VectorIndex fields (name, dimension, metric, vector_count, host) are all optional except name — connectors report what they have.

VectorMatch fields: id, score, metadata: dict, values: list[float] | None.

Building a query

from user_sdk import VectorQueryBuilder, FilterRelation

query = (
    VectorQueryBuilder()
    .embedding([0.1, 0.2, 0.3])
    .top_k(5)
    .filter("genre", FilterRelation.EQ, "drama")
    .filter("year", FilterRelation.GTE, 2020)
    .namespace("prod")
.include_metadata(False) # default True; omit per-match metadata 
    .include_values(True)
    .build()
)
matches = await vec.search(query, index_name="movies")

FilterRelation members: EQ, NEQ, GT, GTE, LT, LTE, IN, NIN. .build() raises ValueError if embedding is empty or top_k <= 0.

Agent client

agent = await client.agent("my-agent-config-id")

# Convenience: natural-language query against a named data source.
answer = await agent.query("How many users signed up last week?", source="my-relational-config-id")

# Or send a raw payload — the agent expects at least "query" and "source".
answer = await agent.invoke({"query": "...", "source": "..."})

Both methods return the agent's response payload as a dict (not the raw MQ envelope). timeout can be overridden per call.

MCP bridge client

McpClient (from client.mcp(...)) proxies requests to an external MCP server registered as a CorpHub MCP service:

mcp = await client.mcp("my-mcp-config-id")

caps = await mcp.get_capabilities()
result = await mcp.call_tool("search_docs", {"query": "vector index"})
resource = await mcp.read_resource("corphub://docs/readme")
prompt = await mcp.get_prompt("summarize", {"length": "short"})

Every method returns the service's response payload as a dict, and each accepts an optional per-call timeout override. Note that when constructed directly, McpClient defaults to a 60s request timeout (MCP tool calls can be slow), whereas client.mcp(config_id, timeout=30) passes 30 like the other factories — pass an explicit timeout= if you need the longer window through the factory.

Semantic mapper client

mapper = await client.mapper("my-mapper-config-id")

ontologies = await mapper.list_ontologies()          # list[str] of filenames
content = await mapper.get_ontology(ontologies[0])   # raw OWL/RDF string

mappings = await mapper.request_mapping("sales-db-connector", ontology="sales.owl")
for m in mappings:                                    # list[FieldMapping]
    print(m.source, "->", m.destination)

ontology= is optional — omit it to let the mapper consider every loaded ontology.

Provisioner client

The provisioner creates a database, applies DDL, creates a scoped app user, and registers a writable-connector config on the hub. It runs asynchronously: create() returns a job_id immediately.

provisioner = await client.provisioner("my-provisioner-config-id")

spec = {
    "cloud_provider": "aws",
    "engine": "postgres",
    "database_name": "sales_summary",
    "config_id": "sales-summary-writable",
    "schema": {...},
}

# Low-level: poll yourself.
started = await provisioner.create(spec)
status = await provisioner.get_status(started["job_id"])

# Convenience: create + poll until terminal, with an overall deadline.
result = await provisioner.provision_and_wait(spec, poll_interval=2.0, timeout=120.0)
print(result["config_id"], result["host"], result["port"], result["tables_created"])

provision_and_wait raises CorphubRequestError if the job fails, expires server-side, or the overall timeout (seconds) elapses — note this bounds the whole poll loop, while each individual create/get_status call still uses the handle's per-request timeout.

Ontology API (tenant-scoped)

Ontologies aren't connector-scoped, so ontology_api() takes no config_id — tenant isolation is enforced server-side from the auth token. This routes through the orchestrator (like every other connector-style call above); se the direct REST path on data_catalog.

ont_api = await client.ontology_api()

ontologies = await ont_api.list(domain="sales")      # list[ Ontology]
one = await ont_api.get(ontologies[0].id)
latest = await ont_api.get_content(one.id)            # OntologyVersion, latest
v2 = await ont_api.get_version(one.id, version=2)
all_versions = await ont_api.get_versions(one.id)

from user_sdk import OntologyFormat

registration = await ont_api.register(
    name="sales-ontology",
    domain="sales",
    format=OntologyFormat.TTL,
    content=turtle_text,
    description="Core sales domain concepts",
)

await ont_api.publish(registration.ontology.id)       # draft -> published
await ont_api.deprecate(registration.ontology.id)     # published -> deprecated

Registration behavior depends on the current state of (name, domain):

• No existing ontology → creates it in draft state as v1.
• Existing draft → overwrites v1 in place (mutable scratch space).
• Existing published → appends a new immutable version.
• Existing deprecated → server rejects the call with CorphubRequestError.

The server also rejects content that's RDF-isomorphic to the latest version and payloads over 100 MB. State transitions are forward-only (draft → published → deprecated); publish()/deprecate() are convenience wrappers around transition_state().

Low-level REST clients

Beyond the connector-style handles above, CorphubClient exposes four REST clients already configured with the gateway URL and your auth token. These can also be constructed standalone if you don't need the full CorphubClient.

RegistryApi (client.registry)

Service and configuration management:

# Services (paginated). type= filters by service type, name= is a
# case-insensitive substring match on the service name (both optional).
services = await client.registry.list_services(
    page=1, page_size=50, type="CONNECTOR", name="sales",
)                                                                          # PagedResult[ServiceInfo]
service = await client.registry.get_service("sales-db-connector")          # ServiceInfo

# Configurations (paginated). config_id= is an optional case-insensitive
# substring filter; total/total_pages reflect the filtered set.
configs = await client.registry.list_configurations(
    page=1, page_size=50, config_id="prod",
)                                                                          # PagedResult[ServiceConfiguration]
by_service = await client.registry.list_configurations_by_service(
    "sales-db-connector", config_id="prod",
)                                                                          # list[ServiceConfiguration]
config = await client.registry.get_configuration("sales-db-connector", "sales-db-connector-prod")
config = await client.registry.get_configuration_by_id("sales-db-connector-prod")  # auto-discovers service

# Create / transition / delete.
result = await client.registry.create_configuration(
    "sales-db-connector", my_service_configuration,
)                                                                          # ConfigurationResult
await client.registry.transition_configuration_state("sales-db-connector-prod", ConfigState.DEPRECATED)
await client.registry.deprecate_configuration("sales-db-connector-prod")     # ready -> deprecated (wrapper)
await client.registry.mark_configuration_ready("sales-db-connector-prod")    # deprecated -> ready (wrapper)
await client.registry.delete_configuration("sales-db-connector", "sales-db-connector-prod")
await client.registry.delete_configuration_by_id("sales-db-connector-prod")  # auto-discovers service

• deprecate_configuration / mark_configuration_ready are thin wrappers over transition_configuration_state(config_id, state), which accepts a ConfigState member or its raw string value.
• create_configuration returns ConfigurationResult(config_id, service_name, message); the other CRUD calls return the affected ServiceConfiguration.
• list_services / list_configurations are genuinely paginated on the server — CorphubClient's factories walk every page rather than trusting the first, so don't assume a single page of 50/100 is exhaustive if you call RegistryApi yourself. Non-2xx responses raise RegistryError; a 404 raises ConfigurationNotFoundError.

SecGateApi (client.sec_gate)

RBAC administration and identity lookups — roles, permissions, users, service tokens, projects, connectors, tenants, and ad-hoc authorization checks:

# Identity
me = await client.sec_gate.get_me()

# Roles
roles = await client.sec_gate.list_roles()
role = await client.sec_gate.create_role({"name": "analyst", ...})
await client.sec_gate.delete_role("role-id")

# Permissions (list filters are all optional)
perms = await client.sec_gate.list_permissions(subject_type="role", subject_id="analyst")
await client.sec_gate.create_permission({"subject_type": "role", "subject_id": "analyst", ...})
await client.sec_gate.delete_permission("perm-id")

# Users
users = await client.sec_gate.list_users()
await client.sec_gate.update_user("user-id", {"role": "admin"})

# Service tokens — create/regenerate return the plaintext token ONCE.
tokens = await client.sec_gate.list_service_tokens()
created = await client.sec_gate.create_service_token({"name": "ci-bot", ...})
await client.sec_gate.update_service_token("token-id", {"name": "ci"})
regenerated = await client.sec_gate.regenerate_service_token("token-id", expires_in_days=90)
await client.sec_gate.delete_service_token("token-id")

# Hierarchy: projects, connectors, tenants
projects = await client.sec_gate.list_projects()
await client.sec_gate.create_project({"name": "sales", ...})
await client.sec_gate.delete_project("project-id")
connectors = await client.sec_gate.list_connectors(project_id="project-id")   # project_id optional
await client.sec_gate.create_connector({"name": "sales-db", ...})
await client.sec_gate.delete_connector("connector-id")
tenants = await client.sec_gate.list_tenants()

# Ad-hoc authorization check
verdict = await client.sec_gate.evaluate({"subject": "...", "action": "read", "resource": "..."})

All methods return raw dict/list[dict] — sec-gate never encrypts its REST responses, so there's no decryption layer to configure. create_service_token and regenerate_service_token are the only calls that ever return a token's plaintext secret (in the response token field); store it immediately.

DataCatalogApi (client.data_catalog)

Direct REST access to connector schema metadata and ontology storage:

# Connector schemas
schemas = await client.data_catalog.list_schemas()
schema = await client.data_catalog.get_schema("sales-db-connector")           # latest version
await client.data_catalog.update_metadata("sales-db-connector", {"owner": "sales-team"})
await client.data_catalog.refresh("sales-db-connector")        # re-pull schema from the live connector

# Ontologies (name= and domain= filters optional)
ontologies = await client.data_catalog.list_ontologies(domain="sales")
ontology = await client.data_catalog.get_ontology("ontology-id")
versions = await client.data_catalog.list_ontology_versions("ontology-id")
one_version = await client.data_catalog.get_ontology_version("ontology-id", version=2)
uploaded = await client.data_catalog.upload_ontology(
    name="sales-ontology", domain="sales",
    content_bytes=turtle_bytes, filename="sales.ttl",
    content_type="text/turtle", format="ttl",
    description="Core sales domain concepts",   # optional
)
await client.data_catalog.transition_ontology_state("ontology-id", "published")

All methods return raw dict / list[dict]. refresh uses a 45s timeout to absorb the live-connector round-trip; upload_ontology sends multipart form data.

This is a separate access path from ontology_api() above — both operate on the same tenant-scoped ontology store, but DataCatalogApi talks REST directly to data-catalog (useful for multipart file uploads), while OntologyApi goes through the orchestrator like every other connector call. Use whichever fits your call site; there's no functional difference for data already registered.

AuditApi (client.audit)

Read access to the immutable audit log, reached through the gateway's /audit reverse proxy:

page = await client.audit.query_events(
    event_type="REQUEST_AUTHORIZED",
    security_decision="GRANTED",
    from_timestamp="2026-03-17T00:00:00Z",
    limit=100,
)
print(page["total"], page["events"])          # newest-first; total is pre-pagination

stats = await client.audit.stats()
print(stats["total_events"], stats["consumer_running"])

Typed models returned by these clients

The registry/schema clients return typed dataclasses (importable from user_sdk unless noted):

• ServiceInfo — name, service_types: list[str] , summary, online, label, image, and config_params: list[ConfigParam]. (ConfigParam itself is not exported from user_sdk.)
• ServiceConfiguration — config_id, service_name, config_values, allowed_roles, offline, description, state (a ConfigState string value), and the registry-computed created_at / updated_at timestamps (empty strings when the registry doesn't report them).
• ConfigState — a StrEnum with READY ("ready") and DEPRECATED ("deprecated"); state toggles are informational and do not stop routing.
• ConfigurationResult — config_id, service_name, message; returned by create_configuration.
• PagedResult[T] — items: list[T], page, page_size, total, total_pages. Returned by list_services / list_configurations. Not exported from user_sdk; treat it as a structural return type rather than something to import.

Tenant selection

A user may belong to more than one tenant. Every request the platform serves is scoped to a single active tenant, resolved server-side by the security gate. When no active tenant is selected, it falls back to the caller's primary (first) tenant.

The active tenant is selected by tenant name via the X-Tenant header. Each low-level REST client (RegistryApi, SecGateApi, DataCatalogApi, AuditApi) accepts an optional tenant= constructor argument that sets that header, and the gateway transport forwards it too:

from user_sdk import RegistryApi

# Target a non-primary tenant by NAME.
registry = RegistryApi(
    "http://localhost:8600/registry",
    auth_token="...",
    tenant="acme-eu",          # -> X-Tenant: acme-eu
)
configs = await registry.list_configurations()   # scoped to the "acme-eu" tenant

Blank/omitted tenant > the header is not sent > the caller's primary tenant is used.

• Limitation: CorphubClient does not currently accept or forward a tenant. The clients it exposes (client.registry, client.sec_gate, client.data_catalog, client.audit) and its connector data-plane calls therefore always run against the primary tenant. To operate on a non-primary tenant today, construct the specific REST client standalone with tenant= as shown above.

Spark integration

Requires the spark extra (pip install "corphub-user-sdk[spark]") and PySpark ≥ 3.5. Each Spark partition maps to one page request sent by that executor directly to the connector — data never passes through the driver.

from pyspark.sql import SparkSession
from user_sdk.spark import configure_spark_session, register_datasource

spark = configure_spark_session(
    SparkSession.builder.appName("my-app"),
    auth_token="...",
    role="analyst",
    gateway_url="http://localhost:8600",
).getOrCreate()
register_datasource(spark)

# Read
orders = spark.read.corphub("sales-db-connector")
pending = spark.read.corphub(
    "sales-db-connector",
    query="SELECT * FROM orders WHERE status = 'pending'",
)

# Write — target config must be a WritableConnector.
summary.write.corphub("sales-db-writable", entity="customer_summary")

configure_spark_session stores corphub.* entries in spark.conf so every subsequent .corphub() call on that session picks up the connection settings automatically, and registers the .corphub() extension methods on DataFrameReader/DataFrameWriter. register_datasource(spark) must be called once per SparkSession before spark.read.corphub(...) works.

Or use the low-level DataFrameReader.format("corphub") API directly, which accepts these options: config_id (required), query, entity, schema (DDL string, overrides auto-detection), auth_token, role, timeout, partition_size (default 5000), gateway_url.

Schema is inferred, in priority order: an explicit schema DDL string → the connector's entity field types (via get_schema()) → a probe of the first row of query's result set. SQL types are mapped to Spark types on a best-effort basis; unrecognized types fall back to StringType.

Pretty-printing schemas

For quick inspection in a notebook or script:

from user_sdk import pretty_print_graphs

graphs = await ds.get_schema()
print(pretty_print_graphs(graphs))

Renders a tree view of every graph, its entities (marking [PK]/[FK] fields), and relations.

Errors

All SDK errors derive from CorphubError, so a single except CorphubError catches everything. Notable subtypes:

The Mq* names are historical — the SDK holds no message-broker connection. MqTimeoutError and MqConnectionError describe the SDK > gateway HTTP hop (see Transport); the gateway performs the internal MQ round-trip on your behalf.

CorphubRequestError exposes machine-readable .code (e.g. "ACCESS_DENIED", "CONNECTOR_ERROR", "ERROR") and optional .status, so you can branch on the failure kind instead of string-matching the message.

from user_sdk import CorphubError, AccessDeniedError, CorphubRequestError

try:
    matches = await vec.search(query, index_name="movies")
except AccessDeniedError:
    ...  # authz denial — don't retry
except CorphubRequestError as e:
    print(e.code, e.status, e)
except CorphubError:
    ...  # catch-all for anything else SDK-originated

Configuration

The SDK reads configuration from constructor arguments and environment variables (12-factor style). Key variables:

The data-plane (connector calls) goes over the gateway and relies on TLS, so it needs no encryption key. CORPHUB_ENCRYPTION_KEY is still honoured by the registry REST client when the registry encrypts its responses.

Samples

Two runnable example scripts ship with the package under examples/. Their key flows are shown below.

consume_connector.py — schema inspection, streaming, filtered query, write-back

import asyncio
from user_sdk.client import CorphubClient


async def main() -> None:
    async with CorphubClient(
        gateway_url="http://localhost:8600",
        auth_token="my-jwt-token",
        role="analyst",
    ) as client:
        # Inspect the schema of a configured connector.
        ds = await client.relational_connector("sales-db-connector")
        for graph in await ds.get_schema():
            for entity in graph.entities:
                print(entity.name, [f.name for f in entity.fields])

        # Stream a full table with auto-pagination + prefetch.
        rows = ds.execute_query("SELECT * FROM orders", page_size=200)
        count = 0
        async for _row in rows:
            count += 1
        print("rows:", count, "reported total:", rows.total_count)
        await rows.close()

        # Write back through a writable connector.
        wds = await client.writable_connector("sales-db-writable")
        result = await wds.write_batch(
            "orders", [{"customer_id": 101, "amount": 299.99, "status": "pending"}]
        )
        print(result.rows_affected, result.status)


asyncio.run(main())

spark_etl.py — Spark read > aggregate > write-back

from pyspark.sql import SparkSession
from user_sdk.spark import configure_spark_session, register_datasource

spark = configure_spark_session(
    SparkSession.builder.appName("sales-etl"),
    auth_token="my-jwt-token",
    role="analyst",
    gateway_url="http://localhost:8600",
).getOrCreate()
register_datasource(spark)

orders = spark.read.corphub("sales-db-connector", query="SELECT * FROM orders")
summary = orders.groupBy("customer_id").sum("amount")

# Persist locally and/or write back to a writable connector.
summary.write.parquet("/tmp/customer_summary.parquet")
summary.write.corphub("sales-db-writable", entity="customer_summary")

Development

pdm install -d       # install with dev/lint/test dependencies
pdm run test         # pytest tests/ -v
pdm run lint         # ruff check src/ tests/
pdm run lint-fix      # ruff check --fix src/ tests/
pdm run format       # ruff format src/ tests/

Availability

Since version 2026-07.