SegmentStream pipeline SDK

Declare source tables with s.ingestion and warehouse SQL transformations with s.model. The SDK resolves inputs, validates explicit partitioning and capabilities, and compiles native Dagster definitions using the existing warehouse loader.

from collections.abc import Iterator

import ibis.expr.types as ir
import segmentstream as s


@s.ingestion(
    tier=s.Tier.Bronze,
    partitioning=s.Partitioning.Unpartitioned,
    schema={"amount": "float64"},
)
def orders() -> Iterator[s.IngestionRow]:
    yield {"amount": 25.0}


@s.model(
    tier=s.Tier.Gold,
    partitioning=s.Partitioning.Unpartitioned,
    schema={"total_revenue": "float64"},
)
def revenue(orders: ir.Table) -> ir.Table:
    return orders.aggregate(total_revenue=orders.amount.sum())


defs = s.definitions(
    assets=[orders, revenue],
    jobs=[s.job("revenue", assets=[revenue], include_upstream=True)],
)

The workspace documentation is organized by concept: connections, ingestion, models, partitioning, jobs, execution, checks, runtime values, and warehouse storage. The CLI bundles these same pages into every starter workspace.

s.validate(defs) checks the graph and named job configuration without executing assets. s.describe(defs) explains resolved inputs, per-asset policies, storage, capabilities, same-day input mappings, and job selections without resolving credentials. See the modular commerce starter for daily row ingestion, batched models, and a partitioned check.

Daily inputs may begin earlier or end later than their consumers when they use the same timezone and day boundaries. The upstream must cover every downstream date. The SDK supplies strict same-day mappings for inputs and ordering-only dependencies; no native Dagster decorator or mapping option is needed. Named jobs still select one daily calendar, with other calendars left as inputs.

Identity-resolution support (0.1.0a39)

Models can declare s.AllPartitions(asset) inputs for retained-history rollups. WarehouseResource.materialize_scratch and iterate execute bounded fixed-point algorithms in warehouse SQL and retain their results until output publication. The SDK owns scratch cleanup; each algorithm owns its convergence and key invariants.

Compatibility

The SDK contains authoring, graph inspection, Ibis resources, connection HTTP clients, and managed secret access. Managed orchestration, workers, PostgreSQL setup, and the metadata webserver live in the internal pipeline runtime.

In 0.1.0a28, use s.get_connection(key).get_http_client() for provider requests. The client forwards requests through the backend, where provider credentials and refresh remain. The access-token and full-credential accessors have been removed. See connections for declaring API destinations.

Workspace pipelines declare segmentstream-pipeline[bigquery]; our image supplies the runtime. The old SDK cloud-run and webserver extras and segmentstream.runtime module are removed in a22.

segmentstream.dagster retains its native exports. Native assets and high-level declarations can coexist in s.definitions. Root s.resource and s.IngestionResource remain available for legacy ingestion. Import their helpers from segmentstream; the old public segmentstream.ingestion module is retired so s.ingestion can be the decorator. Migration examples cover the mapping and IO defaults.

Managed Google clients

The Sandbox runtime integration supplies refreshable credentials to the built-in BigQuery warehouse and GCS IO manager. Ibis remains the user-facing warehouse API; its internal official Google client renews credentials automatically throughout the authorized run. Ten minutes is the lifetime of each access token, not a limit on pipeline duration. Managed execution does not provide general ADC support. See runtime lifecycle and credential brokering. Managed Dagster execution uses the backend compute provider API for every step, including SQL assets and checks. The coordinator owns scheduling and retries; the backend owns execution identity, scoped credentials and cleanup. See the compute provider contract.

Releases

SDK 0.1.0a39 adds explicit all-partition model inputs and managed warehouse iteration with temporary SQL tables, supporting full-history identity resolution.

SDK 0.1.0a38 adds compatible daily calendars with strict same-day dependency mappings. Consumers can start later than their upstream assets; each job still selects one daily calendar.

The GA4 query client targets 0.1.0a37. A model constructs it with s.ga4_bigquery_client(key, context=context) and submits SQL through the backend; BigQuery writes the result directly to the model's warehouse partition. See GA4 BigQuery connections. Publish the SDK before releasing a CLI whose starter pins it. PyPI versions are immutable; do not republish an existing version.

The protected pipeline-sdk-release.yml workflow builds a wheel and source distribution from a pipeline-sdk-v<version> tag, then uses PyPI Trusted Publishing from the pypi GitHub environment. No long-lived publishing token is stored in GitHub.

Unreleased: full-refresh storage layout

Models with s.Partitioning.Unpartitioned may set storage=s.TableStorage(partition_by_date="session_date"). They run as global steps and atomically replace the whole table while retaining physical BigQuery DATE partitions. Daily models keep selected-date replacement. s.describe reports the storage layout independently of execution partitioning. See the model contract. Available in SDK 0.1.0a40; update the workspace dependency and lockfile before using it.

Semantic models (0.1.0a41)

segmentstream.semantics defines portable dimensions, measures, and expressions over declared warehouse assets. Register models with s.definitions(semantic_models=...). Inspection publishes manifest version 8 without adding jobs or materializations. See semantic models. Upgrade backend readers and the internal runtime before deploying definitions from this SDK.

Evaluation connection (0.1.0a43)

EvaluationConnection(key="evaluation") evaluates state against typed boolean, choice and score questions through Vercel AI Gateway's evaluation API; SegmentStream chooses the model and holds the credential, so it needs no configuration or authorization. Call s.evaluation_client("evaluation").evaluate(state=..., questions=...) in managed runs. See evaluation connections. Connections inspection publishes connections manifest schema 8: upgrade the backend before deploying declarations from this SDK.

API-key connections (0.1.0a45)

ApiKeyConnection(key=..., name=..., provider=..., headers={...}, query={...}, apis={...}) declares HTTP access authenticated by static values such as API keys or tokens. Each header or query parameter maps to a Secret(...) configured after deployment, or to secret_ref(...). The backend attaches the values to every proxied request; the connection needs no authorization. Use s.get_connection(key).get_http_client() as with OAuth connections. See connections. Connections inspection publishes connections manifest schema 9: upgrade the backend before deploying declarations from this SDK.

Evaluation refusals (0.1.0a44)

When the evaluation service refuses requests outright, evaluate raises RuntimeContextRefusedError, a Dagster Failure that disallows retries: the step fails at once with the request reference instead of retrying. Earlier SDKs report it as temporary unavailability. See evaluation errors.

HubSpot connection (0.1.0a42)

HubSpotConnection(key="hubspot") declares read-only access to one HubSpot account's CRM through SegmentStream's HubSpot app. Deploy, authorize, and use s.get_connection("hubspot").get_http_client(); the default objects API reads contacts, companies and deals. See connections. Upgrade the backend before deploying declarations from this SDK.