Step¶

Steps are the building blocks of apps. Each step can take inputs, process data, and display results.

Step Class¶

Step ¶

Step(title: str, description: str, parent: str, type: StepType, page: Page, allow_filters: bool = False, uses_pyspark: bool = False, uses_snowflake: bool = False, requires_gpu: bool = False, overrides_action: bool = False, await_actions: bool = True, override_skips_default: bool = False, alerts: Optional[List] = None, default_max_memory_usage: int = None, agent: Optional[DispatcherAgentInterface] = None, show_title: bool = True, *args, **kwargs)

A Step is the basic unit of an app. Steps can be chained together to form an app.

Parameters:

title (str) –

The title of the step.
description (str) –

A description of what the step does.
parent (str) –

The parent step.
type (StepType) –

The Step type.
page (Page) –

The initial Page for this step.
allow_filters (bool, default: False ) –

Whether to allow filters, defaults to False.
uses_pyspark (bool, default: False ) –

Whether or not the step uses PySpark, defaults to False.
uses_snowflake (bool, default: False ) –

Whether or not the step requires a Snowflake connection, defaults to False.
requires_gpu (bool, default: False ) –

Whether this step's compute should be metered as GPU (vs. CPU). Purely declarative — does not allocate or check for a GPU. The metering layer reads this to tag the compute_step event's compute_type. Defaults to False.
overrides_action (bool, default: False ) –

Whether this step overrides the default action for this step type, defaults to False.
await_actions (bool, default: True ) –

whether this step overrides must wait for an action to continue to the next page, defaults to True.
override_skips_default (bool, default: False ) –

When true and overrides_action is true, this flag defines whether to skip over the default step action, defaults to False.
alerts (Optional[List], default: None ) –

The Alerts for this Step. This feature is currently unimplemented, defaults to None.
default_max_memory_usage (int, default: None ) –

The default maximum memory usage of this step in bytes
agent (Optional[DispatcherAgentInterface], default: None ) –

An optional DispatcherAgentInterface for custom LLM behavior
show_title (bool, default: True ) –

Whether to show the title on the page when rendered, defaults to True.

Raises:

PredictException –

Raises an error if the character '/' is found in the flow title.

run `abstractmethod` ¶

run(flow_metadata: FlowMetadata, spark_session=None, *args, **kwargs)

copy ¶

copy()

Step Types¶

StepType ¶

The type of Step being created. INPUT: A Step should be of type input if it contains input elements. DASHBOARD: Marking steps as Dashboard Steps helps them be easily found in the Dashboards section. RESULTS: If a step contains neither inputs or dashboards it should be a Results step.

INPUT `class-attribute` `instance-attribute` ¶

INPUT = 0

DATA_LAB `class-attribute` `instance-attribute` ¶

DATA_LAB = 1

RESULTS `class-attribute` `instance-attribute` ¶

RESULTS = 2

DASHBOARD `class-attribute` `instance-attribute` ¶

DASHBOARD = 3

Creating a Step¶

Steps are created by subclassing the Step class and implementing the run() method:

from virtualitics_sdk import Step, StepType, Page, Section, Card, Table
import pandas as pd

class MyDataStep(Step):
    def run(self, flow_metadata):
        # Your data processing logic
        data = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})

        # Store data for next step
        self._outLink.data = data

        # Return a page to display
        return Page(
            title="My Data",
            sections=[
                Section(
                    title="Results",
                    cards=[Card(title="Table", content=[Table(data=data)])]
                )
            ]
        )

# Instantiate the step
my_step = MyDataStep(
    title="Data Processing",
    description="Processes and displays data",
    parent="Analysis",
    type=StepType.RESULTS,
    page=Page(title="Loading...", sections=[])
)

Step Lifecycle¶

Initialization: Step is created with metadata
run(): Called when step first executes, returns initial Page
action(): Called when user interacts with inputs (optional)
Data Flow: Data passes via _inLink and _outLink

Key Methods¶

run()¶

Required method that executes when the step runs. Must return a Page.

def run(self, flow_metadata):
    # Access data from previous step
    previous_data = self._inLink.data

    # Process data
    result = process_data(previous_data)

    # Store for next step
    self._outLink.result = result

    # Return UI
    return Page(...)

action()¶

Optional method called when user interacts with input elements. Must return a Page.

def action(self, flow_metadata):
    # Get user input
    user_choice = self.page.get_element_by_id("my_dropdown").value

    # Update based on input
    filtered_data = self._inLink.data[self._inLink.data['type'] == user_choice]

    # Return updated UI
    return Page(...)

Data Flow¶

Input Link¶

Access data from the previous step via self._inLink:

def run(self, flow_metadata):
    dataset = self._inLink.dataset
    model = self._inLink.model

Output Link¶

Store data for the next step via self._outLink:

def run(self, flow_metadata):
    self._outLink.processed_data = result
    self._outLink.metadata = {"status": "complete"}

Step Types Explained¶

INPUT: Steps that collect user input (forms, uploads, etc.)
DATA_LAB: Steps for data exploration and preparation
RESULTS: Steps that display results and outputs
DASHBOARD: Steps containing dashboards and visualizations

Advanced Features¶

Progress Tracking¶

Update progress during long-running operations:

def run(self, flow_metadata):
    for i in range(100):
        self._progress = i
        self._message = f"Processing item {i}"
        # ... do work

Memory Management¶

Set memory limits for resource-intensive steps:

my_step = MyStep(
    title="Big Data Processing",
    description="Processes large datasets",
    parent="Analysis",
    type=StepType.RESULTS,
    page=Page(...),
    default_max_memory_usage=4 * 1024**3  # 4GB
)

PySpark Integration¶

Enable PySpark for distributed processing:

my_step = MyStep(
    title="Spark Processing",
    description="Uses PySpark",
    parent="Analysis",
    type=StepType.RESULTS,
    page=Page(...),
    uses_pyspark=True,
    pyspark_config={"spark.executor.memory": "2g"}
)

def run(self, flow_metadata):
    # Access Spark session
    spark_df = self.spark_session.createDataFrame(data)

Best Practices¶

Single Responsibility: Each step should do one thing well
Clear Naming: Use descriptive titles that appear in the UI
Error Handling: Validate inputs and handle edge cases
Progress Updates: Keep users informed during long operations
Type Hints: Use type hints for better code clarity
Docstrings: Document what your step does