Metadata-Version: 2.5
Name: segmentstream-pipeline
Version: 0.1.0a10
Summary: Dagster and Ibis runtime integration for SegmentStream pipelines
Project-URL: Homepage, https://segmentstream.com
Author: SegmentStream
License-Expression: Apache-2.0
License-File: LICENSE
License-File: NOTICE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: <3.14,>=3.12
Requires-Dist: dagster-postgres==0.29.16
Requires-Dist: dagster==1.13.16
Requires-Dist: ibis-framework<13,>=12
Requires-Dist: pydantic<3,>=2.10
Provides-Extra: bigquery
Requires-Dist: ibis-framework[bigquery]<13,>=12; extra == 'bigquery'
Description-Content-Type: text/markdown

# 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.

```python
# 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:

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

```python
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:

```python
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:

```shell
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.

```python
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:

```python
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)
```

```python
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:

```python
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:

```toml
[project]
dependencies = [
  "segmentstream-pipeline[bigquery]==0.1.0a10",
]
```

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.
