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. Application code always uses the same two SDK functions; the SDK selects the credential transport from the execution environment.

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

Managed Dagster assets declare their requirements as metadata. Deployment inspection records the declarations in the pipeline manifest and computes each job's union of capabilities:

import segmentstream.dagster as dg

from segmentstream import runtime_requirements


@dg.asset(
    metadata=runtime_requirements(
        connections=("google_ads",),
        secrets=("google-ads-developer-token",),
    )
)
def google_ads_campaigns() -> object: ...

When that job runs in Cloud Run, the SDK lazily requests each value from the SegmentStream runtime broker. The request must carry both a Google-signed workload identity token and the run's signed, capability-scoped grant. The broker rechecks the active run, deployment, job, service account, and declared requirements before reading the exact stored versions. Values are cached only in the pipeline process after first use. Managed runs do not use the local file descriptor protocol, and pipeline code does not authenticate to Secret Manager or know which warehouse or secret provider is configured.

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. For local execution, possession of the inherited runtime descriptor is the process capability and the CLI retains the user session. For managed execution, Cloud Run supplies the workload identity and SegmentStream supplies the run grant. The local context is read once and closed immediately; managed values are fetched on first use. 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.

Streaming ingestion

API extracts can return a lazy IngestionResource through the same warehouse IO manager used by Ibis assets. Provider code declares a portable Ibis schema, optional primary key, and schema-evolution policy; it does not import or configure a warehouse destination.

from collections.abc import Iterator
from datetime import date

import ibis
import segmentstream.dagster as dg

from segmentstream import (
    IngestionResource,
    SchemaEvolution,
    WAREHOUSE_IO_MANAGER_KEY,
    resource,
    warehouse_asset_metadata,
)


schema = ibis.schema(
    {
        "campaign_id": "!string",
        "metric_date": "!date",
        "clicks": "!int64",
        "raw_payload": "json",
    }
)


@resource(
    schema=schema,
    primary_key=("campaign_id", "metric_date"),
    schema_evolution=SchemaEvolution.ADDITIVE,
)
def campaign_metrics(metric_date: date) -> Iterator[dict[str, object]]:
    # An API iterator can yield any number of rows here.
    yield {
        "campaign_id": "42",
        "metric_date": metric_date,
        "clicks": 7,
        "raw_payload": {"providerField": "preserved"},
    }


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


@dg.asset(
    key=["bronze", "campaign_metrics"],
    partitions_def=daily,
    metadata=warehouse_asset_metadata(partition_by_date="metric_date"),
    io_manager_key=WAREHOUSE_IO_MANAGER_KEY,
    retry_policy=dg.RetryPolicy(
        max_retries=3,
        delay=10,
        backoff=dg.Backoff.EXPONENTIAL,
        jitter=dg.Jitter.PLUS_MINUS,
    ),
)
def raw_campaign_metrics(
    context: dg.AssetExecutionContext,
) -> IngestionResource:
    return campaign_metrics(date.fromisoformat(context.partition_key))

The IO manager consumes the resource once and validates values against the declared schema while writing bounded NDJSON files to an ephemeral local directory. It then loads the files into an isolated staging relation, verifies the extracted row count and primary-key constraints, and commits only after all validation succeeds. An unpartitioned asset replaces its complete table; a daily asset atomically replaces its assigned half-open date range. A successful empty resource is authoritative and clears the corresponding table or date range.

Each spool file defaults to 64 MiB and can be changed with WarehouseResource(ingestion_file_max_bytes=...). Python's standard TMPDIR setting controls the ephemeral spool volume; files are removed after either a successful commit or a failure.

The v1 portable type vocabulary is boolean, signed integer (normalized to int64), floating point (normalized to float64), exact decimal with explicit precision and scale, string, date, UTC timestamp, and JSON. Nested provider records should be kept in a JSON column when they do not deserve stable typed columns. Arrays and structs are intentionally not normalized into child tables. Rows with unknown columns, invalid values, naive timestamps, or missing required columns fail before the target is changed.

SchemaEvolution.ADDITIVE permits only new nullable columns. Column removal, renaming, type changes, nullability changes, and new required columns fail with the target untouched. SchemaEvolution.FREEZE rejects all schema differences. The first protocol intentionally supports only full replacement and daily date partition replacement; append, merge/upsert, arbitrary nested normalization, and source cursor storage remain outside its contract.

Downstream assets use a normal AssetIn dependency. Their input is loaded from the materialized warehouse table as an Ibis expression, so the source rows and NDJSON files do not remain in process memory:

import ibis.expr.types as ir


@dg.asset(
    key=["silver", "active_campaign_metrics"],
    ins={
        "metrics": dg.AssetIn(key=["bronze", "campaign_metrics"]),
    },
    partitions_def=daily,
    metadata=warehouse_asset_metadata(partition_by_date="metric_date"),
    io_manager_key=WAREHOUSE_IO_MANAGER_KEY,
)
def active_campaign_metrics(metrics: ir.Table) -> ir.Table:
    return metrics.filter(metrics.clicks > 0)

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,cloud-run,webserver]==0.1.0a14",
]

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: runtime definition validation, 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.

Every managed execution uses one native asset backfill for the selected job's assets, including selections containing only unpartitioned assets. Daily assets receive the requested inclusive date range; unpartitioned assets remain targets without partition keys. Dagster's coordinator runs its daemon threads in the main runtime process, sharing the open instance. The runtime waits for the backfill's terminal result while checking daemon health and workspace freshness; the controller context manages daemon shutdown. Dagster still owns the separate code-server and child-run worker processes. Re-entering the same public run reuses its persistent backfill. Warehouse IO remains a per-asset decision and does not select the orchestration path.

Child runs use Dagster's implicit asset jobs. The runtime forwards the selected job's run tags and explicit run_config; job-level retry policies, hooks, check exclusions, and per-partition configuration are not inherited by those implicit jobs. Asset-level retry policies remain Dagster-owned. These job-setting limitations also apply to unpartitioned managed executions.

Managed execution installs the SDK's Cloud Run executor on the implicit asset jobs. Every execution step, including warehouse transformations, ingestion, and asset checks, runs in a separate Cloud Run execution. A retry starts a new execution for that step. Dagster's run worker stays in the coordinator container and controls dependencies, retries, concurrency (100 steps per run by default), and completion by reading the shared PostgreSQL event log. Step workers invoke Dagster's native step entrypoint and never start another backfill coordinator.

The shared Dagster run queue permits up to 100 concurrent runs. This is separate from both the per-run step limit and the asset backfill policy: a policy of one partition per run still permits different dates to execute concurrently. Run workers share the coordinator container's CPU and memory, and actual execution concurrency remains subject to available resources and Cloud Run quotas.

Deployment creates one worker Job named segmentstream-step-<deployment UUID without hyphens>, pinned to the deployment image with 1 vCPU, 2 GiB memory, one task per execution, a 24-hour task timeout, and Cloud Run retries disabled. Resource profiles and mixed local/remote step execution are not configured in this version. Older worker Jobs remain available for runs using their deployment. The runtime identity receives execution, cancellation, and read access on its worker Job, plus bucket read access on the existing workspace staging bucket.

The worker receives the public run ID, deployment ID, broker coordinates, and the same scoped credential grant. Warehouse configuration and the Dagster database secret are bound to the Job at deployment. Resolved credentials and database passwords are never copied into step arguments or Dagster run tags. Worker execution references are saved in run tags; a repeated launch reattaches to the same attempt. A lost launch response is recovered from the execution's attempt ID without blindly resubmitting jobs.run. Explicit HTTP 429 launch rejections retry the same step attempt with exponential backoff and jitter, up to eight requests with waits capped at 30 seconds each. Other launch errors are not automatically resubmitted. An unresolved launch or exhausted rate-limit retry budget fails the run. Health checks detect workers that exit without recording a terminal Dagster event. Cancellation requests propagate to Cloud Run; coordinator interruption also cancels its unfinished runs and requests worker cancellation.

Warehouse asset outputs continue to use warehouse_io. For ordinary Python outputs, the managed code location supplies Dagster's GCS pickle IO manager as the default io_manager, using dagster/storage in the workspace staging bucket. Python outputs must be pickleable; custom IO managers must make their outputs accessible to other containers. This does not alter local execution's IO manager or executor. Install the cloud-run extra in managed pipeline images.

Pipeline images that install the webserver extra can also start Dagster's UI and GraphQL API in read-only mode:

python -m segmentstream.runtime serve

The service listens on PORT (default 8080) and loads the workspace's definitions.py. It never starts a Dagster daemon, so SegmentStream remains responsible for scheduling and execution. The code-server loader enriches native Dagster metadata with SDK partition settings, job backfill policies, and multi-cron schedules. GraphQL serves the graph and PostgreSQL-backed execution summaries directly; Cloud Build does not produce a pipeline manifest. The legacy local inspect command remains available, but is not used by deployments.

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.