Schema¶

The Schema class defines and validates data structures.

Schema Class¶

Schema ¶

Schema(schema: Optional[Series] = None, exact_match: bool = False, valid_inputs: Optional[Dict[str, List]] = None, label: Optional[str] = None, metadata: Optional[dict] = None, name: Optional[str] = None, description: Optional[str] = None, version: Optional[int] = None, **kwargs)

A schema asset allows validation of a DataFrame according to a pre-specified schema. A schema is specified using a pd.Series object with indices names as expected columns in the dataframe and values as their expected dtypes. There are 2 levels of validation that can be performed. The first "level 1" validation ensures that the dataframe contains all the expected columns specified in the schema, and optionally ensures that no additional columns are present. It also ensures that each column's dtype matches the schema. The optional "level 2" validation performs additional checks, ensuring that numerical columns are within a specified range of values and that categorical columns take on a value from a specified list. In the event that a check fails, an exception is raised. Otherwise the validation function returns true.

Parameters:

schema (Optional[Series], default: None ) –

A schema object. Indices should be expected column names and values should be expected dtypes.
exact_match (bool, default: False ) –

If true, the dataframe must not have any additional columns not expected in the schema or else an exception will be raised. If performing level 2 validation, the 'valid_inputs' variable must have key/value entries for each column in the schema. If false, these checks will be ignored.
valid_inputs (Optional[Dict[str, List]], default: None ) –

A dictionary mapping column names (str) to lists describing their valid inputs. For columns with a numerical dtype, the value is expected to be [min, max] where min and max is the minimum and maximum possible values in the column respectively. For columns with a "object" dtype (i.e. string/categorical columns) the value is expected to be a list of all possible values in the column. If valid_inputs is None, this level 2 validation will not be performed.
label (Optional[str], default: None ) –

Label for :class:~virtualitics_sdk.assets.asset.Asset, see its documentation for more details.
metadata (Optional[dict], default: None ) –

Metadata for :class:~virtualitics_sdk.assets.asset.Asset, see its documentation for more details.
name (Optional[str], default: None ) –

Name for :class:~virtualitics_sdk.assets.asset.Asset, see its documentation for more details.
description (Optional[str], default: None ) –

Description of :class:~virtualitics_sdk.assets.asset.Asset, see its documentation for more details.
version (Optional[int], default: None ) –

Version for :class:~virtualitics_sdk.assets.asset.Asset, see its documentation for more details.

Basic Usage¶

from virtualitics_sdk import Schema

# Define a schema
schema = Schema(
    name="User Schema",
    schema={
        "username": {"type": "string", "required": True},
        "email": {"type": "string", "required": True},
        "age": {"type": "integer", "minimum": 18},
        "role": {"type": "string", "enum": ["admin", "user", "viewer"]}
    }
)

# Store for validation in other steps
self._outLink.user_schema = schema

Validating Data¶

def run(self, flow_metadata):
    schema = self._inLink.user_schema
    user_data = {
        "username": "john_doe",
        "email": "john@example.com",
        "age": 25,
        "role": "user"
    }

    # Validate against schema
    is_valid = schema.validate(user_data)

    if not is_valid:
        errors = schema.get_validation_errors(user_data)
        # Handle validation errors
        return error_page(errors)

    # Proceed with valid data
    process_user(user_data)

Schema Definition¶

Define schemas using JSON Schema syntax:

product_schema = Schema(
    name="Product Schema",
    schema={
        "product_id": {
            "type": "string",
            "required": True,
            "pattern": "^PROD-[0-9]{6}$"
        },
        "name": {
            "type": "string",
            "required": True,
            "minLength": 3,
            "maxLength": 100
        },
        "price": {
            "type": "number",
            "required": True,
            "minimum": 0,
            "exclusiveMinimum": True
        },
        "category": {
            "type": "string",
            "enum": ["electronics", "clothing", "food", "other"]
        },
        "tags": {
            "type": "array",
            "items": {"type": "string"},
            "uniqueItems": True
        },
        "metadata": {
            "type": "object",
            "properties": {
                "weight": {"type": "number"},
                "dimensions": {
                    "type": "object",
                    "properties": {
                        "length": {"type": "number"},
                        "width": {"type": "number"},
                        "height": {"type": "number"}
                    }
                }
            }
        }
    }
)

DataFrame Validation¶

Validate pandas DataFrames:

# Define column schema
column_schema = Schema(
    name="Sales Data Schema",
    schema={
        "date": {"type": "datetime", "required": True},
        "product": {"type": "string", "required": True},
        "quantity": {"type": "integer", "minimum": 0, "required": True},
        "revenue": {"type": "number", "minimum": 0, "required": True},
        "region": {
            "type": "string",
            "enum": ["North", "South", "East", "West"],
            "required": True
        }
    }
)

# Validate DataFrame
def run(self, flow_metadata):
    df = self._inLink.sales_data.data
    schema = self._inLink.column_schema

    # Check all columns exist
    missing_cols = schema.validate_columns(df.columns)
    if missing_cols:
        return error_page(f"Missing columns: {missing_cols}")

    # Validate data types
    type_errors = schema.validate_types(df)
    if type_errors:
        return error_page(f"Type errors: {type_errors}")

    # Validate value constraints
    validation_errors = schema.validate_rows(df)
    if validation_errors:
        return error_page(f"Validation errors: {validation_errors}")

    # Data is valid, proceed
    process_data(df)

Common Schema Patterns¶

User Input Validation¶

form_schema = Schema(
    name="Contact Form",
    schema={
        "name": {
            "type": "string",
            "required": True,
            "minLength": 2
        },
        "email": {
            "type": "string",
            "required": True,
            "format": "email"
        },
        "phone": {
            "type": "string",
            "pattern": "^\\+?[1-9]\\d{1,14}$"
        },
        "message": {
            "type": "string",
            "required": True,
            "minLength": 10,
            "maxLength": 1000
        }
    }
)

API Response Validation¶

api_schema = Schema(
    name="API Response",
    schema={
        "status": {
            "type": "string",
            "enum": ["success", "error"]
        },
        "data": {
            "type": "object"
        },
        "errors": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "code": {"type": "string"},
                    "message": {"type": "string"}
                }
            }
        }
    }
)

Best Practices¶

Clear Names: Use descriptive schema names
Required Fields: Mark required fields explicitly
Constraints: Define min/max, patterns, and enums where appropriate
Documentation: Add descriptions to schema fields
Reusability: Create reusable schemas for common data structures
Error Handling: Provide clear error messages for validation failures