Metadata-Version: 2.4
Name: rowbase
Version: 0.1.0
Summary: Rowbase SDK — declare data pipelines as Python functions
Author-email: Rowbase Team <team@rowbase.com>
License-Expression: LicenseRef-Proprietary
Keywords: data,etl,pipelines,polars,sdk
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: Other/Proprietary License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: httpx>=0.27.0
Requires-Dist: polars>=1.0
Requires-Dist: pyarrow>=15.0
Requires-Dist: pydantic>=2.0
Requires-Dist: pyyaml>=6.0
Requires-Dist: rich>=13.0
Requires-Dist: typer>=0.15.0
Provides-Extra: dev
Requires-Dist: mypy>=1.13.0; extra == 'dev'
Requires-Dist: pytest>=8.3.0; extra == 'dev'
Requires-Dist: ruff>=0.8.0; extra == 'dev'
Description-Content-Type: text/markdown

# Rowbase

Declare data pipelines as Python functions. Rowbase handles DAG construction, dependency resolution, execution, and validation — so you can focus on your transforms.

## Install

```bash
pip install rowbase
```

## Quick Start

```python
import polars as pl
import rowbase

@rowbase.pipeline
def my_pipeline():
    orders = rowbase.source("orders", columns=["id", "email", "total", "country"])

    @rowbase.dataset("cleaned", data_from=orders)
    def cleaned(orders: pl.DataFrame) -> pl.DataFrame:
        return orders.filter(pl.col("email").is_not_null())

    @rowbase.dataset("domestic", data_from=cleaned)
    def domestic(cleaned: pl.DataFrame) -> pl.DataFrame:
        return cleaned.filter(pl.col("country") == "US")

    yield domestic
```

Run it:

```bash
rowbase pipeline run -p pipeline.py -i orders=orders.csv -o ./output/
```

## Core Concepts

### Sources

Declare data inputs with `source()`. Each source maps to a file provided at runtime.

```python
orders = rowbase.source("orders", columns=["id", "email", "total"])
```

Supported formats: CSV, Parquet, Excel. Detected automatically by file extension.

### Datasets

Transform functions decorated with `@dataset`. They consume sources or other datasets and return a Polars DataFrame.

```python
@rowbase.dataset("summary", data_from=[orders, returns])
def summary(orders: pl.DataFrame, returns: pl.DataFrame) -> pl.DataFrame:
    return orders.join(returns, on="order_id", how="left")
```

### Pipelines

Generator functions that wire sources and datasets together. `yield` a dataset to mark it as a published output — non-yielded datasets are intermediate.

```python
@rowbase.pipeline
def my_pipeline():
    raw = rowbase.source("raw")

    @rowbase.dataset("cleaned", data_from=raw)
    def cleaned(raw: pl.DataFrame) -> pl.DataFrame:
        return raw.drop_nulls()

    @rowbase.dataset("aggregated", data_from=cleaned)
    def aggregated(cleaned: pl.DataFrame) -> pl.DataFrame:
        return cleaned.group_by("category").agg(pl.col("amount").sum())

    yield aggregated  # published output
```

### Schema Validation

Validate dataset outputs with Pydantic models.

```python
from pydantic import BaseModel

class OrderSchema(BaseModel):
    id: int
    email: str
    total: float

@rowbase.dataset("validated", data_from=orders, schema=OrderSchema, on_schema_error="skip")
def validated(orders: pl.DataFrame) -> pl.DataFrame:
    return orders
```

`on_schema_error` options: `"fail"` (default), `"skip"`, `"collect"`.

### Configuration

Define config in `rowbase.yaml`:

```yaml
config:
  api_key: secret_key
  threshold: 100
```

Access values in your pipeline:

```python
rowbase.config.get("api_key")
```

Environment variable overrides follow the pattern `ROWBASE_CONFIG_<KEY>`.

## CLI

```
rowbase pipeline validate -p pipeline.py          # Check DAG structure
rowbase pipeline info -p pipeline.py               # Show sources, datasets, and graph
rowbase pipeline dry-run -p pipeline.py -i orders=orders.csv --sample-rows 5
rowbase pipeline run -p pipeline.py -i orders=orders.csv -o ./output/ --output-format parquet
rowbase pipeline deploy -p pipeline.py             # Deploy to Rowbase platform

rowbase dataset test -d cleaned -p pipeline.py -i orders=orders.csv

rowbase data inspect data.csv                      # Inspect file structure

rowbase auth login                                 # Authenticate
rowbase auth status                                # Check auth state

rowbase runs list                                  # View past runs
rowbase runs show <run_id>                         # Run details
rowbase runs download <run_id> -o ./results/       # Download outputs

rowbase init                                       # Initialize a new project
```

## Programmatic Usage

```python
from pathlib import Path
from rowbase.execution import PipelineRunner

spec = my_pipeline()
runner = PipelineRunner()
result = runner.run(
    spec,
    inputs={"orders": Path("orders.csv")},
    output_dir=Path("output/"),
    output_format="parquet",
)

print(result.status)  # "success", "partial", or "failed"
for name, df in result.dataframes.items():
    print(f"{name}: {df.shape[0]} rows")
```

## Output Formats

Parquet, CSV, NDJSON, and Excel. Set via `--output-format` in the CLI or `output_format` in `PipelineRunner.run()`.

## License

Proprietary. All rights reserved.
