rowbase 0.1.1

Rowbase SDK — declare data pipelines as Python functions

Rowbase

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

Install

pip install rowbase

Quick Start

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:

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.

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.

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

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

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:

config:
  api_key: secret_key
  threshold: 100

Access values in your pipeline:

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

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.