Assets¶

Assets are typed data objects that provide structure, validation, and persistence for data flowing through your apps.

Why Use Assets?¶

Instead of storing raw Python objects in Links, assets provide:

Type Safety: Clear data types (Dataset, Model, Schema)
Validation: Ensure data meets requirements
Metadata: Store additional context
Serialization: Automatic persistence
Documentation: Self-documenting code

Asset Types¶

Dataset¶

For tabular data (pandas DataFrames):

from virtualitics_sdk import Dataset
import pandas as pd

df = pd.DataFrame({'col1': [1, 2, 3], 'col2': [4, 5, 6]})

dataset = Dataset(
    name="Sales Data",
    data=df
)

self._outLink.sales = dataset

Model¶

For trained machine learning models:

from virtualitics_sdk import Model
from sklearn.ensemble import RandomForestClassifier

clf = RandomForestClassifier()
clf.fit(X_train, y_train)

model = Model(
    name="Sales Predictor",
    model=clf,
    metadata={
        "accuracy": 0.95,
        "features": list(X_train.columns)
    }
)

self._outLink.trained_model = model

Schema¶

For data validation:

from virtualitics_sdk import Schema

schema = Schema(
    name="User Schema",
    schema={
        "name": {"type": "string", "required": True},
        "age": {"type": "integer", "minimum": 18},
        "email": {"type": "string", "format": "email"}
    }
)

self._outLink.validation_schema = schema

Using Assets¶

Storing Assets¶

def run(self, flow_metadata):
    # Create DataFrame
    df = load_data()

    # Wrap in Dataset
    dataset = Dataset(
        name="Customer Data",
        data=df,
        metadata={
            "source": "database",
            "loaded_at": datetime.now().isoformat()
        }
    )

    # Store in Link
    self._outLink.customer_data = dataset

    return Page(...)

Retrieving Assets¶

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

    # Access the DataFrame
    df = dataset.data

    # Access metadata
    print(f"Data source: {dataset.metadata['source']}")

    # Use the data
    summary = df.describe()

    return Page(...)

Complete Workflow Example¶

# Step 1: Load and store data
class LoadDataStep(Step):
    def run(self, flow_metadata):
        df = pd.read_csv('sales.csv')

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

        self._outLink.sales_data = dataset
        return Page(...)

# Step 2: Train model
class TrainModelStep(Step):
    def run(self, flow_metadata):
        # Get data from previous step
        df = self._inLink.sales_data.data

        # Train model
        X = df[['feature1', 'feature2']]
        y = df['target']
        clf = RandomForestClassifier()
        clf.fit(X, y)

        # Store model
        model = Model(
            name="Sales Predictor",
            model=clf,
            metadata={
                "features": ['feature1', 'feature2'],
                "train_score": clf.score(X, y)
            }
        )

        self._outLink.model = model
        return Page(...)

# Step 3: Make predictions
class PredictStep(Step):
    def run(self, flow_metadata):
        # Get model and new data
        model = self._inLink.model
        new_data = self._inLink.new_data

        # Make predictions
        predictions = model.model.predict(new_data.data)

        # Store results
        results = new_data.data.copy()
        results['prediction'] = predictions

        self._outLink.predictions = Dataset(
            name="Predictions",
            data=results
        )

        return Page(...)

Best Practices¶

Always Use Names: Give descriptive names to assets
Add Metadata: Store context and provenance information
Validate Data: Use Schema assets to validate data structure
Type Hints: Use type hints for clarity
Document Assets: Comment what each asset contains

Metadata Best Practices¶

Include useful metadata:

dataset = Dataset(
    name="Processed Sales",
    data=df,
    metadata={
        "source": "sales_database",
        "processed_at": datetime.now().isoformat(),
        "row_count": len(df),
        "columns": list(df.columns),
        "processing_steps": ["filter", "aggregate", "clean"],
        "version": "2.0"
    }
)