Metadata-Version: 2.4
Name: moncpipelib
Version: 0.41.3
Summary: Shared utilities for Model Oncology data pipelines
Project-URL: Homepage, https://github.com/model-oncology-public/moncpipelib
Project-URL: Repository, https://github.com/model-oncology-public/moncpipelib
Author-email: Model Oncology <engineering@modeloncology.com>
License-Expression: Apache-2.0
License-File: LICENSE
License-File: NOTICE
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: azure-identity>=1.19
Requires-Dist: azure-keyvault-secrets>=4.10
Requires-Dist: azure-storage-blob>=12.23
Requires-Dist: dagster>=1.12.20
Requires-Dist: httpx>=0.28
Requires-Dist: ijson<4,>=3.2
Requires-Dist: polars>=1.39.3
Requires-Dist: psycopg[binary]>=3.1
Requires-Dist: pyyaml>=6.0.3
Requires-Dist: sqlalchemy>=2.0.48
Requires-Dist: sqlglot>=29.0.1
Provides-Extra: dev
Requires-Dist: freezegun>=1.5.0; extra == 'dev'
Requires-Dist: mypy>=1.19.1; extra == 'dev'
Requires-Dist: openlineage-python>=1.45.0; extra == 'dev'
Requires-Dist: pytest-cov>=7.1.0; extra == 'dev'
Requires-Dist: pytest>=9.0.2; extra == 'dev'
Requires-Dist: respx>=0.22.0; extra == 'dev'
Requires-Dist: ruff>=0.15.7; extra == 'dev'
Requires-Dist: types-pyyaml>=6.0.12.20250915; extra == 'dev'
Provides-Extra: integration
Requires-Dist: testcontainers[azurite,postgres]>=4.14.2; extra == 'integration'
Provides-Extra: openlineage
Requires-Dist: openlineage-python>=1.45.0; extra == 'openlineage'
Description-Content-Type: text/markdown

# moncpipelib

moncpipelib is a Dagster-based data-pipeline framework: it provides the
resources, IO managers, contract enforcement, lineage, and ingest patterns that
Dagster code locations compose into pipelines. It owns the I/O boundary policy
(streaming-by-default reads/writes against Postgres and blob), the
contract/lineage policy, and the partition/period coordination model.

It is maintained by Model Oncology and developed against our internal
data-pipeline needs, released under the Apache 2.0 license.

## Features

- **PostgreSQL IO Manager** -- write Polars DataFrames to PostgreSQL with multiple write modes (full refresh, upsert, append, partitioned)
- **PostgreSQL Resource** -- streaming/batched reads for large tables via server-side cursors
- **Row-level lineage tracking** -- UUID7-based lineage with foreign key to a centralized lineage table
- **Data contracts** -- declarative YAML-based schema validation, auto-enforced on write
- **OpenLineage integration** -- emit lineage events to Marquez, DataHub, or other backends
- **Data transforms** -- `clean_text`, `safe_decimal`, `safe_bool`, `safe_date`, and more

## Installation

```bash
uv add moncpipelib
# or
pip install moncpipelib
```

Requires Python 3.11+.

To work from source:

```bash
git clone https://github.com/model-oncology-public/moncpipelib
cd moncpipelib
uv sync --all-extras --dev
```

## Quick Start

```python
from dagster import asset, Definitions, EnvVar
import polars as pl
from moncpipelib import PostgresResource, PostgresIOManager, clean_text, safe_decimal

database = PostgresResource(
    host=EnvVar("DB_HOST"), port=EnvVar.int("DB_PORT"),
    user=EnvVar("DB_USER"), password=EnvVar("DB_PASSWORD"),
    database=EnvVar("DB_NAME"),
)

@asset
def orders_bronze(database: PostgresResource) -> pl.DataFrame:
    return database.read_batched_to_dataframe("SELECT * FROM raw.orders")

@asset(io_manager_key="silver_io_manager")
def orders_silver(orders_bronze: pl.DataFrame) -> pl.DataFrame:
    return orders_bronze.select([
        clean_text("order_id"),
        safe_decimal("amount"),
    ])

defs = Definitions(
    assets=[orders_bronze, orders_silver],
    resources={
        "database": database,
        "silver_io_manager": PostgresIOManager(
            postgres_resource=database,
            default_schema="silver",
        ),
    },
)
```

For more examples, see the auto-generated [Cookbook](docs/cookbook.md).

## Documentation

| Topic | Link |
|-------|------|
| Usage examples (auto-generated) | [docs/cookbook.md](docs/cookbook.md) |
| Database resources and IO managers | [docs/best-practices.md](docs/best-practices.md) |
| Data contracts specification | [docs/data-contracts-spec.md](docs/data-contracts-spec.md) |
| Row-level lineage tracking | [docs/lineage-tracking.md](docs/lineage-tracking.md) |
| OpenLineage integration | [docs/openlineage-integration-spec.md](docs/openlineage-integration-spec.md) |
| SCD Type 2 guide | [docs/scd2-guide.md](docs/scd2-guide.md) |
| Security controls | [docs/security.md](docs/security.md) |

## Configuration

moncpipelib uses environment variables for global configuration:

| Variable | Default | Description |
|----------|---------|-------------|
| `MONCPIPELIB_DEFAULT_DATABASE` | `analytics` | Default database name for contract sources/sinks that omit one |
| `MONCPIPELIB_OPENLINEAGE_SCHEMA_URL` | (repo URL) | Base URL for OpenLineage custom facet schemas |
| `MONCPIPELIB_OPENLINEAGE_NAMESPACE` | `moncpipelib` | Default namespace for OpenLineage jobs/datasets |
| `MONCPIPELIB_LINEAGE_TABLE` | `data_lineage` | Name of the lineage tracking table |
| `MONCPIPELIB_LINEAGE_SCHEMA` | `lineage` | Schema containing the lineage tracking table |

## Development

```bash
uv sync --all-extras         # Install dev dependencies
uv run pytest                # Run tests
uv run ruff check src tests  # Lint
uv run mypy src              # Type check
uv run ruff format src tests # Format
```

See [CONTRIBUTING.md](CONTRIBUTING.md) before opening a pull request.

## License

Apache License 2.0 -- see [LICENSE](LICENSE).
