Skip to content

Data Flow

Understanding how data flows through your app is crucial for building effective workflows.

Data flows between steps via Links. Each step has:

  • Input link (self._inLink): Receives data from the previous step
  • Output link (self._outLink): Sends data to the next step
Step 1 ─[outLink]─→ Step 2 ─[outLink]─→ Step 3
         [inLink]←           [inLink]←

Storing Data

In any step, store data in _outLink:

def run(self, flow_metadata):
    # Create some data
    df = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})

    # Wrap in Dataset
    dataset = Dataset(name="My Data", data=df)

    # Store in outLink
    self._outLink.dataset = dataset
    self._outLink.metadata = {"rows": len(df), "cols": len(df.columns)}

    return Page(...)

Retrieving Data

In the next step, retrieve from _inLink:

def run(self, flow_metadata):
    # Get data from previous step
    dataset = self._inLink.dataset
    metadata = self._inLink.metadata

    # Access the DataFrame
    df = dataset.data

    # Use it
    summary = df.describe()

    return Page(...)

Data Types

You can store any Python object in Links, but using typed assets is recommended:

from virtualitics_sdk import Dataset, Model, Schema

# Dataset for DataFrames
self._outLink.sales_data = Dataset(name="Sales", data=df)

# Model for ML models
self._outLink.trained_model = Model(name="Predictor", model=clf)

# Schema for validation
self._outLink.schema = Schema(name="DataSchema", schema=schema_def)

Using Plain Python Objects

# Also works, but less structured
self._outLink.my_dict = {"key": "value"}
self._outLink.my_list = [1, 2, 3]
self._outLink.my_value = 42

Multi-Step Flow Example

# Step 1: Load Data
class LoadStep(Step):
    def run(self, flow_metadata):
        df = pd.read_csv('data.csv')
        self._outLink.raw_data = Dataset(name="Raw", data=df)
        return Page(...)

# Step 2: Clean Data
class CleanStep(Step):
    def run(self, flow_metadata):
        raw = self._inLink.raw_data.data
        cleaned = raw.dropna()
        self._outLink.cleaned_data = Dataset(name="Cleaned", data=cleaned)
        return Page(...)

# Step 3: Analyze Data
class AnalyzeStep(Step):
    def run(self, flow_metadata):
        data = self._inLink.cleaned_data.data
        analysis = data.describe()
        self._outLink.analysis = Dataset(name="Analysis", data=analysis)
        return Page(...)

Conditional Flow

You can control which data gets passed:

def action(self, flow_metadata):
    user_choice = self.page.get_element_by_id("analysis_type").value

    if user_choice == "statistical":
        self._outLink.analysis_type = "stats"
        self._outLink.data = statistical_analysis(data)
    elif user_choice == "ml":
        self._outLink.analysis_type = "ml"
        self._outLink.data = ml_analysis(data)

    return Page(...)

Data Persistence

Data in Links is automatically persisted:

  • PostgreSQL: For small data and metadata
  • S3: For large datasets, models, and files
  • In-Memory: During active execution

You don't need to manage persistence manually - it's handled automatically.

Best Practices

  1. Use Assets: Wrap data in Dataset, Model, or Schema for type safety
  2. Descriptive Names: Use clear names for stored data
  3. Store Selectively: Only store what's needed by subsequent steps
  4. Validate Data: Check that expected data exists before using
  5. Document Flow: Comment what data each step expects/produces

Data Validation

Check for required data:

def run(self, flow_metadata):
    # Check if required data exists
    if not hasattr(self._inLink, 'dataset'):
        return error_page("No dataset found from previous step")

    # Validate data structure
    df = self._inLink.dataset.data
    required_cols = ['id', 'name', 'value']

    if not all(col in df.columns for col in required_cols):
        return error_page(f"Dataset missing required columns: {required_cols}")

    # Data is valid, proceed
    result = process_data(df)
    return Page(...)

See Also