Schema Management

Advanced schema management capabilities for defining, validating, and maintaining data schemas in ViewAI.

Overview

ViewAI provides comprehensive schema management tools that help you:

• Automatically generate schemas from data

• Define custom validation rules

• Compare and migrate schemas

• Export schemas in multiple formats

• Ensure data consistency across your ML pipeline

Understanding Schemas

A schema in ViewAI defines the structure, types, and constraints of your model's input features. It includes:

• Field Definitions: Data types, ranges, and categories for each feature

• Validation Rules: Constraints for data quality and consistency

• Metadata: Model information, creation timestamps, and statistics

Schema Structure

{
    "model_id": "my_model",
    "fields": {
        "age": {
            "type": "int",
            "min": 0,
            "max": 120
        },
        "income": {
            "type": "float",
            "min": 0.0,
            "max": 1000000.0,
            "mean": 55000.0,
            "std": 25000.0,
            "missing_count": 0
        },
        "region": {
            "type": "category",
            "choices": ["north", "south", "east", "west"],
            "allow_new_categories": False,
            "n_unique": 4,
            "missing_count": 0
        }
    },
    "metadata": {
        "created_at": "2025-10-11T12:00:00",
        "n_features": 3,
        "n_samples": 10000,
        "model_type": "RandomForestClassifier"
    }
}

Automatic Schema Generation

Basic Schema Generation

Generate schemas automatically from your training data:

from viewai_client.models.schema import ModelSchemaManager
import pandas as pd
from sklearn.ensemble import RandomForestClassifier

# Prepare your data
df = pd.DataFrame({
    'age': [25, 30, 35, 40, 45],
    'income': [50000.0, 60000.0, 55000.0, 70000.0, 65000.0],
    'region': ['north', 'south', 'east', 'west', 'north'],
    'status': [0, 1, 0, 1, 1]
})

# Train a model
X = df[['age', 'income', 'region']]
y = df['status']
model = RandomForestClassifier()
model.fit(X, y)

# Initialize schema manager
schema_manager = ModelSchemaManager(
    model_id="customer_churn",
    api_key="your-api-key"
)

# Generate schema from data
schema = schema_manager.generate_schema_from_dataframe(
    model=model,
    df=df
)

print(f"Generated schema with {len(schema['fields'])} fields")

Controlling Category Handling

Control how categorical features handle new values:

# Allow new categories for specific fields
schema = schema_manager.generate_schema_from_dataframe(
    model=model,
    df=df,
    allowed_categories=['region', 'country'],  # Only these can have new values
    always_allow_categories=False
)

# Allow new categories for all categorical fields
schema = schema_manager.generate_schema_from_dataframe(
    model=model,
    df=df,
    always_allow_categories=True
)

Custom Schema Definition

Defining Schemas Manually

Create schemas from scratch for precise control:

schema_manager = ModelSchemaManager(api_key="your-api-key")

# Define custom schema
custom_schema = {
    "model_id": "risk_assessment",
    "fields": {
        "credit_score": {
            "type": "int",
            "min": 300,
            "max": 850
        },
        "annual_income": {
            "type": "float",
            "min": 0.0,
            "max": 10000000.0,
            "mean": 75000.0,
            "std": 35000.0
        },
        "employment_status": {
            "type": "category",
            "choices": ["employed", "self-employed", "unemployed", "retired"],
            "allow_new_categories": False
        },
        "loan_purpose": {
            "type": "category",
            "choices": ["home", "auto", "education", "business"],
            "allow_new_categories": True
        }
    },
    "metadata": {
        "created_at": pd.Timestamp.now().isoformat(),
        "n_features": 4,
        "description": "Risk assessment model schema"
    }
}

# Set the schema
schema_manager.schema = custom_schema

Updating Schema Fields

Modify individual schema fields after creation:

# Update numeric range
schema_manager.update_column_range('credit_score', min_value=300, max_value=900)

# Update categorical choices
schema_manager.update_column_categories(
    'employment_status',
    categories=['employed', 'self-employed', 'unemployed', 'retired', 'student'],
    allow_new_categories=False
)

# Change data type
schema_manager.update_column_data_type('user_id', 'str')

Schema Validation

Using SchemaValidator

Validate schema structure and data compliance:

from viewai_client.utils.schema_utils import SchemaValidator

validator = SchemaValidator()

# Validate schema structure
valid, errors = validator.validate_schema_structure(schema)

if not valid:
    print("Schema validation errors:")
    for error in errors:
        print(f"  - {error}")
else:
    print("Schema structure is valid")

Validating Data Against Schema

Ensure your data conforms to the schema:

# Validate DataFrame against schema
result = validator.validate_data_against_schema(
    data=df,
    schema=schema,
    strict=True  # All errors are critical
)

if not result['valid']:
    print(f"Validation failed: {result['summary']}")
    print("\nErrors:")
    for error in result['errors']:
        print(f"  - {error}")

    print("\nWarnings:")
    for warning in result['warnings']:
        print(f"  - {warning}")

# View field-level statistics
for field_name, stats in result['field_stats'].items():
    print(f"\n{field_name}:")
    print(f"  Type: {stats['type']}")
    print(f"  Nulls: {stats['null_count']}")
    print(f"  Unique: {stats['unique_count']}")

    if stats['type'] in ['int', 'float']:
        print(f"  Min: {stats.get('min_value')}")
        print(f"  Max: {stats.get('max_value')}")

Non-Strict Validation

Use non-strict mode for warnings instead of errors:

# Non-strict validation allows extra fields
result = validator.validate_data_against_schema(
    data=df,
    schema=schema,
    strict=False  # Extra fields generate warnings, not errors
)

print(f"Valid: {result['valid']}")
print(f"Errors: {len(result['errors'])}")
print(f"Warnings: {len(result['warnings'])}")

Schema Comparison

Comparing Schemas

Compare two schemas to identify differences:

from viewai_client.utils.schema_utils import SchemaComparator

comparator = SchemaComparator()

# Compare schemas
old_schema = schema_manager.load_schema_from_file('schema_v1.json')
new_schema = schema_manager.load_schema_from_file('schema_v2.json')

diff = comparator.compare_schemas(old_schema, new_schema)

if diff['has_changes']:
    print("Schema Changes Detected:")

    if diff['added_fields']:
        print(f"\nAdded Fields: {diff['added_fields']}")

    if diff['removed_fields']:
        print(f"\nRemoved Fields: {diff['removed_fields']}")

    if diff['modified_fields']:
        print("\nModified Fields:")
        for mod in diff['modified_fields']:
            print(f"  {mod['field']}:")
            print(f"    Old: {mod['old']}")
            print(f"    New: {mod['new']}")

Detecting Breaking Changes

Identify changes that could break existing integrations:

breaking_changes = comparator.detect_breaking_changes(diff)

if breaking_changes:
    print("WARNING: Breaking Changes Detected!")
    for change in breaking_changes:
        print(f"  - {change}")
else:
    print("No breaking changes detected")

Generating Migration Plans

Create migration steps for schema updates:

migration_plan = comparator.generate_migration_plan(diff)

if migration_plan:
    print("Migration Plan:")
    for i, step in enumerate(migration_plan, 1):
        print(f"  {i}. {step}")
else:
    print("No migration needed")

Schema Export and Import

Exporting Schemas

Export to JSON

from viewai_client.utils.schema_utils import SchemaExporter

exporter = SchemaExporter()

# Export schema to JSON file
exporter.export_to_json(
    schema=schema_manager.schema,
    file_path='./schemas/model_schema_v1.json'
)

Export to Markdown Documentation

# Export schema as human-readable documentation
exporter.export_to_markdown(
    schema=schema_manager.schema,
    file_path='./docs/schema_documentation.md'
)

The generated Markdown includes:

• Model ID and metadata

• Field definitions with types and constraints

• Categorical choices and numeric ranges

• Creation timestamps and statistics

Importing Schemas

# Load schema from JSON file
schema = schema_manager.load_schema_from_file('./schemas/model_schema_v1.json')

# Import and validate
imported_schema = exporter.import_from_json('./schemas/model_schema_v1.json')
valid, errors = validator.validate_schema_structure(imported_schema)

if valid:
    schema_manager.schema = imported_schema
    print("Schema imported and validated successfully")

Schema Versioning

Version Control Best Practices

import json
from datetime import datetime

class SchemaVersionManager:
    """Manage schema versions with metadata."""

    def __init__(self, schema_manager):
        self.schema_manager = schema_manager
        self.versions = []

    def save_version(self, version_name, description=""):
        """Save current schema as a version."""
        schema_copy = self.schema_manager.get_schema().copy()

        version_info = {
            "version": version_name,
            "description": description,
            "timestamp": datetime.now().isoformat(),
            "schema": schema_copy
        }

        self.versions.append(version_info)

        # Save to file
        filename = f"schema_{version_name}.json"
        with open(filename, 'w') as f:
            json.dump(version_info, f, indent=2)

        print(f"Schema version {version_name} saved")

    def load_version(self, version_name):
        """Load a specific schema version."""
        filename = f"schema_{version_name}.json"

        with open(filename, 'r') as f:
            version_info = json.load(f)

        self.schema_manager.schema = version_info['schema']
        print(f"Loaded schema version {version_name}")
        return version_info

# Usage
version_manager = SchemaVersionManager(schema_manager)

# Save versions
version_manager.save_version("v1.0.0", "Initial production schema")
version_manager.save_version("v1.1.0", "Added new regional field")

# Load specific version
version_info = version_manager.load_version("v1.0.0")

API Integration

Uploading Schemas to ViewAI

# Upload schema to ViewAI API
response = schema_manager.upload_schema_to_api(
    model_id="customer_churn_v2",
    schema=schema_manager.schema
)

if response:
    print("Schema uploaded successfully")
else:
    print("Schema upload failed")

Fetching Schemas from ViewAI

# Fetch existing schema from API
schema = schema_manager.fetch_schema_from_api("customer_churn_v2")

if schema:
    print(f"Fetched schema for model: {schema.get('model_id')}")
    print(f"Fields: {list(schema.get('fields', {}).keys())}")

Saving Schemas Locally

# Save schema to local file
schema_manager.save_schema_to_file('./schemas/production_schema.json')

# Load schema from local file
schema_manager.schema = schema_manager.load_schema_from_file(
    './schemas/production_schema.json'
)

ModelSchemaManager Reference

Initialization

from viewai_client.models.schema import ModelSchemaManager

# With API key for remote operations
schema_manager = ModelSchemaManager(
    model_id="my_model",
    api_key="your-api-key"
)

# Without API key for local operations
schema_manager = ModelSchemaManager(model_id="my_model")

Core Methods

# Schema Generation
schema = schema_manager.generate_schema_from_dataframe(model, df)

# Validation Rules
rules = schema_manager.get_validation_rules()

# Schema Updates
schema_manager.update_column_range('age', 0, 120)
schema_manager.update_column_categories('region', ['north', 'south'])
schema_manager.update_column_data_type('user_id', 'str')

# File Operations
schema_manager.save_schema_to_file('schema.json')
schema = schema_manager.load_schema_from_file('schema.json')

# API Operations
schema_manager.upload_schema_to_api(model_id, schema)
schema = schema_manager.fetch_schema_from_api(model_id)

# Access Schema
current_schema = schema_manager.get_schema()

Handling Missing Values

Configuring Missing Value Rules

# Define custom missing value handling
missing_rules = {
    'age': 0,  # Fill with 0
    'income': lambda x: 50000 if pd.isna(x) else x,  # Fill with median
    'region': 'unknown',  # Fill with placeholder
}

# Apply missing value rules
df_cleaned = schema_manager.handle_missing_values(
    df=df,
    missing_value_rules=missing_rules
)

# Default handling (fill with "NaN" string)
df_default = schema_manager.handle_missing_values(df=df)

Complete Example

1

Prepare Data

from viewai_client.models.schema import ModelSchemaManager
from viewai_client.utils.schema_utils import (
    SchemaValidator,
    SchemaComparator,
    SchemaExporter
)
import pandas as pd
from sklearn.ensemble import RandomForestClassifier

df = pd.DataFrame({
    'age': [25, 30, 35, 40, 45],
    'income': [50000.0, 60000.0, 55000.0, 70000.0, 65000.0],
    'region': ['north', 'south', 'east', 'west', 'north'],
    'status': [0, 1, 0, 1, 1]
})

2

Train Model

X = df[['age', 'income', 'region']]
y = df['status']
model = RandomForestClassifier()
model.fit(X, y)

3

Generate Schema

schema_manager = ModelSchemaManager(
    model_id="customer_model",
    api_key="your-api-key"
)

schema = schema_manager.generate_schema_from_dataframe(
    model=model,
    df=df,
    allowed_categories=['region']
)

4

Validate Schema

validator = SchemaValidator()
valid, errors = validator.validate_schema_structure(schema)
print(f"Schema valid: {valid}")

5

Validate Data

result = validator.validate_data_against_schema(df, schema, strict=True)
print(f"Data validation: {result['summary']}")

6

Export Schema

exporter = SchemaExporter()
exporter.export_to_json(schema, './schema_v1.json')
exporter.export_to_markdown(schema, './schema_docs.md')

7

Upload to API

response = schema_manager.upload_schema_to_api("customer_model", schema)

8

Modify Schema

schema_manager.update_column_range('age', 18, 100)
schema_manager.update_column_categories(
    'region',
    categories=['north', 'south', 'east', 'west', 'central'],
    allow_new_categories=True
)

9

Save New Version

schema_manager.save_schema_to_file('./schema_v2.json')

10

Compare Versions

comparator = SchemaComparator()
old_schema = exporter.import_from_json('./schema_v1.json')
new_schema = exporter.import_from_json('./schema_v2.json')

diff = comparator.compare_schemas(old_schema, new_schema)
breaking_changes = comparator.detect_breaking_changes(diff)
migration_plan = comparator.generate_migration_plan(diff)

print(f"Breaking changes: {breaking_changes}")
print(f"Migration plan: {migration_plan}")

Best Practices

• Version Control: Always version your schemas and track changes

• Validation: Validate both schema structure and data before deployment

• Documentation: Export schemas to Markdown for team documentation

• Breaking Changes: Check for breaking changes before schema updates

• Backup: Keep backup copies of production schemas

• Testing: Test schema changes with sample data before production

• Migration: Create and follow migration plans for schema updates

• Consistency: Ensure schemas match between training and inference

• Metadata: Include comprehensive metadata for debugging

• Review: Review auto-generated schemas before deployment

Troubleshooting

Schema Upload Fails

# Check API key and model ID
schema_manager = ModelSchemaManager(
    model_id="valid_model_id",
    api_key="valid_api_key"
)

Data Validation Errors

# Use non-strict mode to see warnings
result = validator.validate_data_against_schema(df, schema, strict=False)
print(result['warnings'])

Type Mismatch

# Check data types match schema
for col in df.columns:
    print(f"{col}: {df[col].dtype}")

Missing Fields

# Check for required fields
schema_fields = set(schema['fields'].keys())
data_fields = set(df.columns)
missing = schema_fields - data_fields
print(f"Missing fields: {missing}")

Next Steps

• Learn about Health Monitoring

• Explore Model Registry

• Review Best Practices