SegmentStream pipeline SDK

segmentstream-pipeline contains the Python authoring surface and runtime contract for a SegmentStream workspace. Dagster continues to own assets and jobs, while Ibis continues to own relational expressions. This package supplies declarative connection declarations, lazy runtime configuration, and durable warehouse I/O.

Connections

Connections are workspace-level declarations and live in a separate top-level connections/ component alongside app/ and pipeline/. Each custom OAuth connection is executable by the isolated connection runtime, while importing the declaration performs no authorization, network access, or secret resolution.

# connections/connections.py

from segmentstream.connections import OAuth2Connection, secret_ref


connections = (
    OAuth2Connection(
        key="google_ads",
        name="Google Ads",
        provider="google",
        authorization_url="https://accounts.google.com/o/oauth2/v2/auth",
        token_url="https://oauth2.googleapis.com/token",
        client_id=secret_ref("google-oauth-client-id"),
        client_secret=secret_ref("google-oauth-client-secret"),
        scopes=["https://www.googleapis.com/auth/adwords"],
        authorization_parameters={
            "access_type": "offline",
            "prompt": "consent",
        },
    ),
)

The connection key is the stable identifier used by pipelines, the CLI, and agents; name is its user-facing label. Provider endpoints live in source. Both the client ID and client secret are symbolic references supplied through the control plane and resolved only inside the application's private connection runtime. Callback URLs, authorization codes, refresh tokens, and access tokens are runtime values and never belong in connection declarations. The runtime implements authorization URL construction, authorization-code exchange, and refresh-token exchange; the control plane owns OAuth state and durable encrypted token storage.

Pipeline and discovery code can resolve only values explicitly granted to its execution. For local discovery, the CLI passes a one-use runtime context over an inherited file descriptor; values are not written to the project, command arguments, or environment variables:

segmentstream exec \
  --connection google_ads \
  --secret google-ads-developer-token \
  -- uv run python discover_accounts.py

import segmentstream


oauth = segmentstream.get_connection_credentials("google_ads")
developer_token = segmentstream.get_secret("google-ads-developer-token")

get_connection_credentials returns a provider-neutral OAuth2Credentials value containing the client ID, client secret, access token, refresh token, token URL, expiry, token type, and scopes needed to construct an official provider client. For example, a project that depends on Google's official authentication library can adapt it directly:

from google.oauth2.credentials import Credentials
import segmentstream


oauth = segmentstream.get_connection_credentials("google_ads")
google_credentials = Credentials(
    token=oauth.access_token,
    refresh_token=oauth.refresh_token,
    token_uri=oauth.token_url,
    client_id=oauth.client_id,
    client_secret=oauth.client_secret,
    scopes=list(oauth.scopes),
    # google-auth currently represents UTC expiry as a naive datetime.
    expiry=oauth.expires_at.replace(tzinfo=None) if oauth.expires_at else None,
)

The SegmentStream SDK does not authenticate independently. Possession of the inherited runtime descriptor is the process capability; the CLI retains the user session and the deployed runtime uses its workload identity. The context is read once, closed immediately, and cached only in process memory. Credential fields are excluded from object representations, but Python strings cannot be reliably zeroed, so code receiving these values belongs inside the execution's trust boundary.

Execution credentials are a snapshot. Official clients may refresh them during the local process, but this first protocol does not write a rotated refresh token back to SegmentStream. If a provider rotates refresh tokens, a later execution may require reauthorizing the connection. SegmentStream refreshes credentials near expiry before issuing the snapshot, so short discovery commands normally use the returned access token without refreshing it.

The separately published segmentstream-connections-runtime distribution runs from the workspace's connections/ directory:

python -m segmentstream_connections_runtime inspect manifest.json
python -m segmentstream_connections_runtime serve

The service listens on PORT (default 8080) and exposes private internal authorize, code-exchange, and refresh operations. References are injected as namespaced environment values—for example, google-oauth-client-id maps to SEGMENTSTREAM_CONNECTION_SECRET_GOOGLE_OAUTH_CLIENT_ID. Production Cloud Run services must require IAM authentication; these token-bearing endpoints are not a public workspace API.

The initial connector supports BigQuery, automatic dataset creation, full-table replacement for unpartitioned assets, and native daily DATE partitioning. It reads the following non-secret configuration when a pipeline first accesses the warehouse:

• SEGMENTSTREAM_WAREHOUSE_ENGINE
• SEGMENTSTREAM_WAREHOUSE_CATALOG
• SEGMENTSTREAM_WAREHOUSE_DEFAULT_NAMESPACE
• SEGMENTSTREAM_WAREHOUSE_LOCATION (optional)

Configuration and authentication are deliberately lazy. Importing and validating definitions.py during a deployment build does not connect to a warehouse. In Cloud Run, the BigQuery connector uses the attached workload identity through Application Default Credentials.

import ibis
import ibis.expr.types as ir
import segmentstream.dagster as dg

from segmentstream import WAREHOUSE_IO_MANAGER_KEY, warehouse_resources


BRONZE_ORDERS = dg.AssetKey(["bronze", "orders"])
SILVER_ORDERS = dg.AssetKey(["silver", "orders"])


@dg.asset(
    key=BRONZE_ORDERS,
    io_manager_key=WAREHOUSE_IO_MANAGER_KEY,
    kinds={"ibis"},
)
def orders() -> ir.Table:
    return ibis.memtable(
        [{"order_id": "o-1", "amount": 100.0}],
        schema={"order_id": "string", "amount": "float64"},
    )


@dg.asset(
    key=SILVER_ORDERS,
    ins={"orders": dg.AssetIn(key=BRONZE_ORDERS)},
    io_manager_key=WAREHOUSE_IO_MANAGER_KEY,
    kinds={"ibis"},
)
def normalized_orders(orders: ir.Table) -> ir.Table:
    return orders.filter(orders.amount > 0)


defs = dg.Definitions(
    assets=[orders, normalized_orders],
    resources=warehouse_resources(),
)

Daily assets use Dagster's native daily partitions and declare the physical BigQuery DATE column through SegmentStream metadata:

from datetime import date

import ibis
import ibis.expr.types as ir
import segmentstream.dagster as dg

from segmentstream import (
    WAREHOUSE_IO_MANAGER_KEY,
    warehouse_asset_metadata,
)


daily = dg.DailyPartitionsDefinition(start_date="2026-01-01")


@dg.asset(
    key=["silver", "daily_orders"],
    partitions_def=daily,
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=10),
    metadata=warehouse_asset_metadata(partition_by_date="event_date"),
    io_manager_key=WAREHOUSE_IO_MANAGER_KEY,
)
def daily_orders(context: dg.AssetExecutionContext) -> ir.Table:
    partition_dates = [date.fromisoformat(key) for key in context.partition_keys]
    return ibis.memtable(
        [{"event_date": value, "order_count": 0} for value in partition_dates],
        schema={"event_date": "date", "order_count": "int64"},
    )

SegmentStream accepts only unpartitioned assets and default-midnight DailyPartitionsDefinition assets with YYYY-MM-DD keys. Deployment inspection rejects other partition definitions and daily assets without physical partition metadata. The IO manager maps Dagster's partition time window to a half-open warehouse date range, filters upstream Ibis relations to that range, creates the table with native daily partitioning on first materialization, and atomically replaces only those dates on subsequent materializations.

Asset keys map to relations using a small convention:

• ["orders"] uses the configured default namespace.
• ["bronze", "orders"] uses the explicit bronze dataset.
• Other key shapes are rejected.

The workspace project is always supplied by SegmentStream and cannot be overridden by an asset. Before writing an asset, the IO manager creates its validated dataset with CREATE SCHEMA IF NOT EXISTS in the configured location. This lets pipeline authors organize one workspace project into datasets such as bronze, silver, and gold without provisioning them separately.

For local package development, install this project in editable mode rather than adding a relative path dependency to a deployable pipeline.

Workspace pipelines declare only the SegmentStream SDK. It installs the pinned Dagster and Ibis versions that belong to that SDK release:

[project]
dependencies = [
  "segmentstream-pipeline[bigquery]==0.1.0a9",
]

Pipeline definitions import segmentstream.dagster as their curated Dagster namespace. Its objects are direct re-exports from Dagster, not wrappers. APIs outside that namespace are not part of the SegmentStream Cloud compatibility contract even if they remain importable from the underlying dependency.

The same installed package contains SegmentStream's private Cloud Run runtime: the deployment inspector, persistent Dagster instance setup, and ephemeral backfill coordinator. Workspace code does not call these modules directly. Keeping them in this distribution ensures that the SDK, Dagster, and dagster-postgres versions always move together; the backend only builds and launches the installed runtime.

Releases

Releases use the version declared in pyproject.toml and are published from the pipeline-sdk-v<version> Git tag by the protected pipeline-sdk-release.yml workflow. The workflow builds the wheel and source distribution in a job without publishing credentials, then uses PyPI Trusted Publishing from the pypi GitHub environment. No long-lived PyPI token is stored in GitHub.

PyPI releases are immutable. Increment the package version before creating a new release tag; do not reuse a version that has already been uploaded.