Metadata-Version: 2.5
Name: segmentstream-pipeline
Version: 0.1.0a1
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==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` is the small runtime contract between a workspace's
Dagster definitions and the warehouse provisioned by SegmentStream. Dagster
continues to own assets and jobs, while Ibis continues to own relational
expressions. This package supplies lazy runtime configuration and durable
warehouse I/O.

The initial connector supports BigQuery and full-table replacement. 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.

```python
import dagster as dg
import ibis.expr.types as ir

from segmentstream import external_table_asset, warehouse_resources


raw_orders = external_table_asset(key=["raw", "orders"])


@dg.asset(
    key=["analytics", "normalized_orders"],
    io_manager_key="warehouse_io",
    kinds={"ibis"},
)
def normalized_orders(raw_orders: ir.Table) -> ir.Table:
    return raw_orders.filter((raw_orders.status == "completed") & (raw_orders.amount > 0))


defs = dg.Definitions(
    assets=[raw_orders, normalized_orders],
    resources=warehouse_resources(),
)
```

Asset keys map to relations using a small convention:

- `["orders"]` uses the configured default namespace.
- `["analytics", "orders"]` uses the explicit `analytics` namespace.
- Other key shapes are rejected.

For local package development, install this project in editable mode rather
than adding a relative path dependency to a deployable pipeline.

Published workspace pipelines will install the provider extra:

```sh
python -m pip install "segmentstream-pipeline[bigquery]"
```

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