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, then uses PyPI Trusted Publishing from the pypi GitHub environment. No long-lived publishing token is stored in GitHub. Merging a new version in pyproject.toml to main releases it and then creates its pipeline-sdk-v<version> tag; a version already tagged or on PyPI is skipped. Pushing that tag, or running the workflow manually, also releases the version. A pull request that changes src/ or the published pyproject.toml metadata must bump the version (scripts/check-sdk-versions.mjs); the app and ontology SDKs follow the same rule and release the same way.

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.

Ontology entity sources (0.1.0a56)

Pipeline manifest v10 describes assets, schemas, jobs and checks. Declare business measures in ontology entity sources. The semantic_models argument and segmentstream.semantics package were removed. Release platform support before upgrading workspaces.

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.

Path values and Telegram connection (0.1.0a58)

ApiKeyConnection accepts path={"<prefix>": Secret(...)} for providers that read the value from the URL path: the backend inserts <prefix><value> right after the API's base path. TelegramBotConnection(key="telegram") declares the Telegram Bot API with the token from @BotFather, sent as https://api.telegram.org/bot<token>/. Connections inspection publishes connections manifest schema 14: upgrade the backend before deploying declarations from this SDK. See Telegram.

Managed connection scopes and LinkedIn conversions (0.1.0a57)

Every managed preset accepts scopes= and publishes an explicit array in connections manifest schema 13. Defaults request all available permissions known to this SDK release; pin a non-empty subset to retain it across upgrades. TikTok and X publish empty arrays because their permissions are configured on the application. Existing absent/null declarations retain their previous defaults; upgrading an unchanged LinkedIn Ads declaration adds rw_conversions and requires reauthorization. Pin scopes=("r_ads", "r_ads_reporting") to keep its current authorization. The existing client supports conversion-rule discovery and creation, and single or batch conversion events. Upgrade the backend before deploying schema 13. See managed access and LinkedIn conversions.

Own OAuth applications and LinkedIn Community (0.1.0a55)

LinkedInAdsConnection, HubSpotConnection and GoogleSheetsConnection accept application="own": users enter their own OAuth client ID and secret on the connection and authorize through it, with the managed connector's scopes and read-only operations. LinkedInCommunityConnection(key="linkedin") reads a LinkedIn member's post and follower analytics and the pages they administer through the Community Management API; scopes narrows the requested permissions to those its application has. Connections inspection publishes connections manifest schema 12: upgrade the backend before deploying declarations from this SDK. See connections.

Mercury, Starling and Wise connections (0.1.0a61)

MercuryConnection(key="mercury"), StarlingConnection(key="starling") and WiseConnection(key="wise") declare their bank APIs with Authorization: Bearer <token>. Use a read-only Mercury API token, a Starling personal access token with account:read, balance:read, transaction:read and account-list:read, or a read-only Wise personal API token. These ApiKeyConnection presets publish the existing connections manifest schema 14. See Mercury, Starling and Wise for token creation and account limitations.

Apollo connection (0.1.0a54)

ApolloConnection(key="apollo") declares the Apollo.io API at https://api.apollo.io/api/v1/ with an API key, which the backend sends on the x-api-key header. It is an ApiKeyConnection preset and publishes the existing connections manifest schema 11. See connections.

Supabase connection (0.1.0a53)

SupabaseConnection(key="supabase", project="<ref>") declares one Supabase project's REST API at https://<ref>.supabase.co/rest/v1/ with a secret key (sb_secret_), which the backend sends on the apikey header. It is an ApiKeyConnection preset and publishes the existing connections manifest schema 11. See connections.

Stripe connection (0.1.0a52)

StripeConnection(key="stripe") declares the Stripe API with an API key, preferably a restricted read-only key; the backend sends it as Authorization: Bearer <key>. It is an ApiKeyConnection preset and publishes the existing connections manifest schema 11. See connections.

Client credentials and Xero connection (0.1.0a51)

ClientCredentialsConnection declares OAuth 2.0 client-credentials access: users configure a client ID and secret, and the backend exchanges them for bearer tokens. XeroConnection reads one Xero organisation through a Xero custom connection; declare one key per organisation. Connections inspection publishes connections manifest schema 11: upgrade the backend before deploying declarations from this SDK. See connections.

Bearer values and Resend connection (0.1.0a50)

ApiKeyConnection headers accept Bearer(Secret(...)), so users configure only a token and the backend sends Bearer <token>. ResendConnection(key="resend") declares the Resend API with an API key. Connections inspection publishes connections manifest schema 10: upgrade the backend before deploying declarations from this SDK. See connections.

Apify connection (0.1.0a49)

ApifyConnection(key="apify") declares access to the Apify API with a personal API token. Deploy, paste the token into the connection, and use s.get_connection("apify").get_http_client() with paths relative to https://api.apify.com/v2/. It is an ApiKeyConnection preset, so it publishes the existing connections manifest schema 9. See connections.

Job schedules (0.1.0a48)

s.schedule(name, job=, cron=, timezone=, partitions=s.LatestPartitions(count)) runs a named job on a cron schedule, at most once an hour. Register schedules with s.definitions(schedules=...). SegmentStream's scheduler, not Dagster, starts the runs. Daily jobs run the newest count complete partitions on each tick. Inspection publishes manifest version 9. Native Dagster ScheduleDefinitions now fail validation; they never ran in SegmentStream. Upgrade the backend and the internal runtime before deploying definitions from this SDK.

Google Sheets connection (0.1.0a47)

GoogleSheetsConnection(key="google_sheets") declares read-only access to the spreadsheets one Google account can open, through SegmentStream's Google app. Deploy, authorize, and use s.get_connection("google_sheets").get_http_client() to read spreadsheet metadata and values by spreadsheet ID. See connections. Upgrade the backend before deploying declarations from this SDK.

Local TLS roots (0.1.0a46)

Connection HTTP clients trust the Python installation's CA roots, as managed runtimes require. When that store is empty, as in python.org macOS builds where "Install Certificates.command" was not run, they also trust certifi's bundle, so segmentstream exec works without setting SSL_CERT_FILE. A certificate that still cannot be verified raises RuntimeContextError naming the verification failure, with the original error as its cause; earlier SDKs report it as temporary unavailability.

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.

Event streams

An optional events/ component declares managed collectors separately from the pipeline. Its pyproject.toml must depend on segmentstream-pipeline>=0.1.0a59. Export a tuple named streams from events/events.py:

from segmentstream.events import Stream, auth, protocol

streams = (
    Stream(
        key="website",
        protocol=protocol.Json(),
        auth=auth.WriteKey(),
        max_body_bytes=65536,
        retention_days=None,
    ),
)

Stream keys use lowercase letters, digits, and underscores and begin with a letter. Each request contains one JSON object; the default limit is 64 KiB and may be configured up to 1 MiB. Raw partition retention defaults to no expiry. The platform inspects these declarations in an isolated build and owns keys, collection endpoints, and the append-only bronze.events_website table. Use segmentstream events get website to discover its endpoint, and segmentstream events rotate-key website to receive a new write key.

Read a stream's raw table as an external pipeline source. The physical sink uses a timestamp partition on collector receipt time; an unpartitioned external asset lets models choose their own time filters without replacing the raw sink:

import segmentstream as s

events_website = s.external_table(
    key="events_website",
    tier=s.Tier.Bronze,
    partitioning=s.Partitioning.Unpartitioned,
)

An explicit empty streams = () removes collection routes while retaining history. Deployments preserve operational pauses, and reintroducing a retired stream requires an explicit resume. A sender must keep its event_id across retries: collection and warehouse delivery are at least once, so downstream models deduplicate by stream and event ID.

Declare event payload contracts with native Ibis types:

import ibis.expr.datatypes as dt
from segmentstream.events import EventSchema, Stream

purchase = EventSchema(
    event_type="purchase",
    version=1,
    schema={
        "order_id": dt.String(),
        "amount": dt.Decimal(18, 2),
        "items": dt.Array(dt.Struct({"sku": dt.String(), "quantity": dt.Int64()})),
        "purchased_at": dt.Timestamp("UTC", 6),
    },
)
streams = (Stream(key="website", schemas=(purchase,)),)

schema accepts the same Ibis schema or name-to-type mapping as models. Every node, including array elements, must be nullable; non-nullable declarations are rejected. Payload fields are optional. Decimal values use exact decimal strings, and timestamps use RFC3339 values with a timezone and at most microsecond precision. Supported types are string, boolean, signed integer widths, float64, bounded BigQuery decimals, date, UTC microsecond timestamps, structs and arrays. Arrays remain JSON columns in generated views, preserving nested arrays and null elements.

Add fields by publishing a higher version; preserve all existing names, case, order and types recursively. require_additive(previous, candidate) checks this contract locally, while publication checks cumulative immutable history. Event schemas are snapshotted, hashed deterministically and inspected as declarative validator/view metadata. They never contain executable SQL. Retirement retains historical schemas and view columns. A stream may have up to 64 active schemas, each with 256 named fields, 16 type levels and a 256 KiB artifact. Event schema versions are positive integers up to 9,007,199,254,740,991; root fields reserve the case-insensitive _ss_ prefix.

Send the direct JSON request as an explicit envelope:

{"event_type":"purchase","schema_version":1,"payload":{"order_id":"9081","amount":"49.90"}}

Missing or invalid schema selectors, old flat requests, unknown fields and schema mismatches are retained and flagged. Unknown payload fields are preserved. An HTTP 202 confirms publication acceptance; validation status separately reports whether supplied fields passed. Schemas require pipeline SDK 0.1.0a60 or later.