Metadata-Version: 2.5
Name: segmentstream-pipeline
Version: 0.1.0a14
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'
Provides-Extra: cloud-run
Requires-Dist: dagster-gcp==0.29.16; extra == 'cloud-run'
Provides-Extra: webserver
Requires-Dist: dagster-webserver==1.13.16; extra == 'webserver'
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. 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:

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

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:

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

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

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

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