Tutorial: Working with Data¶

Learn how to load, transform, and persist data across steps in Virtualitics apps.

Overview¶

This tutorial covers:

Loading data from CSV, databases, and DataSource elements
Using Dataset assets for structured data flow
Transforming and aggregating data between steps
Showing progress for long operations
Displaying data in tables and charts

Prerequisites¶

Completed Your First App tutorial
Basic pandas knowledge

Loading Data¶

From Inline / Generated Data¶

The simplest approach — create a DataFrame directly in your step:

import pandas as pd
from virtualitics_sdk import Step, Dataset

class GenerateDataStep(Step):
    def run(self, flow_metadata):
        df = pd.DataFrame({
            "date": pd.date_range("2025-01-01", periods=30, freq="D"),
            "sales": [100 + i * 5 for i in range(30)],
        })

        self._outLink.data = Dataset(name="Generated Data", data=df)
        return Page(...)

From CSV Files¶

Load CSV files bundled with your app or accessible from the worker filesystem:

class LoadCSVStep(Step):
    def run(self, flow_metadata):
        df = pd.read_csv("data/sales.csv")

        dataset = Dataset(
            name="Sales Data",
            data=df,
            metadata={
                "source": "sales.csv",
                "rows": len(df),
                "columns": list(df.columns),
            }
        )

        self._outLink.raw_data = dataset

        return Page(
            title="Data Loaded",
            sections=[
                Section(
                    title="Raw Data",
                    cards=[
                        Card(
                            title=f"Loaded {len(df)} rows",
                            content=[Table(data=df)]
                        )
                    ]
                )
            ]
        )

From a Database via DataSource¶

Use the DataSource element to let users pick from configured database connections:

from virtualitics_sdk import DataSource

class ConnectStep(Step):
    def run(self, flow_metadata):
        datasource = DataSource(
            id="db_source",
            title="Select Database",
            connection_type="postgresql"
        )

        return Page(
            title="Connect",
            sections=[
                Section(
                    title="Data Source",
                    cards=[Card(title="Database", content=[datasource])]
                )
            ]
        )

    def action(self, flow_metadata):
        ds = self.page.get_element_by_id("db_source")
        # Use the selected connection to query data
        df = pd.read_sql("SELECT * FROM orders LIMIT 1000", ds.connection)
        self._outLink.orders = Dataset(name="Orders", data=df)
        return Page(...)

Transforming Data Between Steps¶

Each step receives data from the previous step via _inLink and passes results forward via _outLink.

class CleanStep(Step):
    """Remove nulls and add computed columns."""

    def run(self, flow_metadata):
        raw = self._inLink.raw_data.data

        # Drop rows with missing values
        df = raw.dropna()

        # Add computed column
        df["total"] = df["quantity"] * df["unit_price"]

        # Convert types
        df["date"] = pd.to_datetime(df["date"])

        self._outLink.cleaned = Dataset(name="Cleaned Data", data=df)

        return Page(
            title="Cleaned",
            sections=[
                Section(
                    title=f"Kept {len(df)} of {len(raw)} rows",
                    cards=[Card(title="Preview", content=[Table(data=df.head(20))])]
                )
            ]
        )


class AggregateStep(Step):
    """Aggregate by category."""

    def run(self, flow_metadata):
        df = self._inLink.cleaned.data

        summary = (
            df.groupby("category")
            .agg(total_revenue=("total", "sum"),
                 order_count=("total", "count"),
                 avg_price=("unit_price", "mean"))
            .round(2)
            .reset_index()
        )

        self._outLink.summary = Dataset(name="Summary", data=summary)

        return Page(
            title="Summary",
            sections=[
                Section(
                    title="Revenue by Category",
                    cards=[Card(title="Aggregated", content=[Table(data=summary)])]
                )
            ]
        )

Showing Progress for Long Operations¶

For steps that take a while, update _progress (0-100) and _message to keep the user informed:

class ProcessStep(Step):
    def run(self, flow_metadata):
        df = self._inLink.raw_data.data
        results = []

        for i, (_, row) in enumerate(df.iterrows()):
            self._progress = int((i / len(df)) * 100)
            self._message = f"Processing row {i + 1} of {len(df)}"

            result = expensive_computation(row)
            results.append(result)

        result_df = pd.DataFrame(results)
        self._outLink.results = Dataset(name="Results", data=result_df)

        return Page(...)

You can also use the SDK's tqdm wrapper for automatic progress updates:

from virtualitics_sdk.utils.tqdm import StepProgressTqdm

class BatchStep(Step):
    def run(self, flow_metadata):
        items = self._inLink.items

        with StepProgressTqdm(self, total=len(items), desc="Processing") as pbar:
            for item in items:
                process(item)
                pbar.update(1)

        return Page(...)

Storing Multiple Assets¶

You can store any number of named assets in _outLink:

def run(self, flow_metadata):
    # Store multiple datasets
    self._outLink.train_data = Dataset(name="Training", data=train_df)
    self._outLink.test_data = Dataset(name="Testing", data=test_df)

    # Store a model
    from virtualitics_sdk import Model
    self._outLink.model = Model(name="Classifier", model=trained_clf)

    # Store plain Python objects
    self._outLink.config = {"threshold": 0.5, "features": feature_list}

    return Page(...)

In the next step, retrieve them by the same attribute names:

def run(self, flow_metadata):
    train = self._inLink.train_data.data
    test = self._inLink.test_data.data
    model = self._inLink.model.model
    config = self._inLink.config

Validating Data¶

Check that required data exists and meets expectations:

def run(self, flow_metadata):
    # Check existence
    if not hasattr(self._inLink, "dataset"):
        return Page(
            title="Error",
            sections=[
                Section(title="Error", cards=[
                    Card(title="Missing Data", content=[
                        RichText("No dataset found from the previous step.")
                    ])
                ])
            ]
        )

    df = self._inLink.dataset.data

    # Check required columns
    required = ["id", "name", "value"]
    missing = [c for c in required if c not in df.columns]
    if missing:
        return Page(
            title="Error",
            sections=[
                Section(title="Validation Error", cards=[
                    Card(title="Missing Columns", content=[
                        RichText(f"Dataset is missing columns: {missing}")
                    ])
                ])
            ]
        )

    # Data is valid — proceed
    return process_and_display(df)

Complete Multi-Step Example¶

Putting it all together — a three-step pipeline:

from virtualitics_sdk import (
    App, Step, StepType, Page, Section, Card,
    Table, PlotlyPlot, Dataset
)
import pandas as pd
import plotly.express as px


class LoadStep(Step):
    def run(self, flow_metadata):
        df = pd.DataFrame({
            "product": ["A", "B", "C", "A", "B", "C"],
            "quarter": ["Q1", "Q1", "Q1", "Q2", "Q2", "Q2"],
            "revenue": [120, 95, 200, 140, 110, 180],
        })
        self._outLink.data = Dataset(name="Revenue", data=df)
        return Page(
            title="Loaded",
            sections=[Section(title="Raw", cards=[
                Card(title="Data", content=[Table(data=df)])
            ])]
        )


class AnalyzeStep(Step):
    def run(self, flow_metadata):
        df = self._inLink.data.data
        summary = df.groupby("product")["revenue"].sum().reset_index()
        self._outLink.summary = Dataset(name="Summary", data=summary)
        return Page(
            title="Analysis",
            sections=[Section(title="By Product", cards=[
                Card(title="Summary", content=[Table(data=summary)])
            ])]
        )


class VisualizeStep(Step):
    def run(self, flow_metadata):
        summary = self._inLink.summary.data
        fig = px.bar(summary, x="product", y="revenue", title="Revenue by Product")
        return Page(
            title="Results",
            sections=[Section(title="Chart", cards=[
                Card(title="Revenue", content=[PlotlyPlot(figure=fig)])
            ])]
        )


load = LoadStep(title="Load", description="Load data",
                parent="Data", type=StepType.INPUT,
                page=Page(title="Loading...", sections=[]))

analyze = AnalyzeStep(title="Analyze", description="Aggregate",
                      parent="Analysis", type=StepType.DATA_LAB,
                      page=Page(title="Loading...", sections=[]))

visualize = VisualizeStep(title="Visualize", description="Chart",
                          parent="Results", type=StepType.DASHBOARD,
                          page=Page(title="Loading...", sections=[]))

app = App(name="Revenue Pipeline", description="Load → Analyze → Visualize")
app.chain([load, analyze, visualize])