rowbase 0.3.0

Rowbase SDK — declare data pipelines as Python functions

Rowbase SDK

Status: Alpha
Python: 3.12+
Core Dependencies: Polars, Pydantic, Typer

---

Overview

Rowbase SDK is a Python library for Rowbase engineers to author data pipelines. These pipelines are then deployed and made available to non-technical customers who simply upload their data and receive cleaned results.

Key positioning:

• Agentic authoring — Rowbase engineers write pipelines (often with AI assistance)
• Invisible to users — Customers never see the code; they just upload files and get results
• B2B data cleaning — Customers are non-technical business users who need clean data

Who Uses This

User	Use Case
Rowbase Engineers	Write pipelines using the SDK, deploy via CLI
End Customers	Upload data via web UI, download cleaned results

The customer never interacts with the SDK directly.

Installation

pip install rowbase
# or: uv add rowbase

Quick Start

import rowbase
import polars as pl

@rowbase.pipeline
def orders_etl():
    orders = rowbase.source("orders", columns=["order_id", "customer", "amount", "status"])
    customers = rowbase.source("customers", columns=["customer_id", "name", "email"])

    @rowbase.dataset("completed_orders", data_from=orders)
    def filter_completed(df):
        return df.filter(pl.col("status") == "completed")

    @rowbase.dataset("enriched_orders", data_from=[filter_completed, customers])
    def enrich(orders_df, customers_df):
        return orders_df.join(
            customers_df,
            left_on="customer",
            right_on="customer_id",
        )

    yield completed_orders
    yield enriched_orders

Authoring Workflow

1. Understand customer data — What does their raw data look like?
2. Define sources — What columns/format do they upload?
3. Write transformations — Use Polars to clean/transform
4. Test locally — Use rowbase run to test
5. Deploy — Use rowbase deploy to deploy to the platform
6. Customer uses — Customer uploads files via web UI

Core Concepts

@rowbase.pipeline

Decorator that marks a function as a pipeline. The function should be a generator that yields published datasets.

@rowbase.pipeline
def my_pipeline():
    # ... sources and datasets ...
    yield published_dataset

rowbase.source()

Declares an input data source. Returns a SourceHandle used as input to datasets.

orders = rowbase.source(
    "orders",
    columns=["order_id", "customer", "amount"],
    description="Raw orders from Shopify",
    reader_options={"separator": ",", "has_header": True},
    optional=False,
)

@rowbase.dataset

Declares a transformation function. The decorated function receives DataFrames and returns a DataFrame.

@rowbase.dataset(
    "cleaned_data",
    data_from=source_handle,
    schema=MyPydanticModel,
    on_schema_error="fail",
    description="Cleaned and validated data",
    metadata=True,
)
def transform(df):
    return df.filter(pl.col("amount") > 0)

CLI Commands

rowbase --help

# Initialize a new pipeline project
rowbase init --name my-pipeline

# Run locally for testing
rowbase run

# Validate before deploying
rowbase validate

# Deploy pipeline to Rowbase
rowbase deploy

# List deployed pipelines
rowbase list

# Inspect local data files
rowbase data inspect data/input.csv

Deployment

To deploy a pipeline:

rowbase deploy

This validates your pipeline locally, uploads the code to the Rowbase API, and creates a new version that customers can use. rowbase push also works as an alias.

Customer Experience

Once deployed, customers see the pipeline in their dashboard:

1. Select pipeline — Choose "Orders ETL" from the list
2. Upload files — Upload orders.csv and customers.csv
3. Submit — Click "Run"
4. Wait — See progress (pending → running → completed)
5. Download — Get enriched_orders.csv with clean data

Example: Data Cleaning Pipeline

import rowbase
import polars as pl

@rowbase.pipeline
def clean_customer_data():
    customers = rowbase.source("customers")

    @rowbase.dataset("valid_emails", data_from=customers)
    def filter_valid(df):
        return df.filter(pl.col("email").str.contains("@"))

    @rowbase.dataset("normalized_phones", data_from=valid_emails)
    def normalize_phones(df):
        return df.with_columns(
            pl.col("phone").str.replace_all(r"[^0-9]", "").alias("phone")
        )

    yield valid_emails
    yield normalized_phones

Dependencies

• polars - DataFrame operations
• pydantic - Schema validation
• typer - CLI framework
• pyarrow - Parquet support
• xlsxwriter - Excel support
• fastexcel - Fast Excel parsing

License

Proprietary - All rights reserved