Services

This document provides comprehensive documentation for all service classes in the ViewAI Python SDK. Services follow a service-oriented architecture pattern, encapsulating specific domains of functionality.

Overview

The ViewAI SDK provides specialized service classes for different operations:

• PredictionService: Execute predictions and manage inference jobs

• ModelTrainingService: Train and manage machine learning models

• WorkspaceManager: Manage workspace resources

• ProjectManager: Manage project resources

• HealthChecker: Monitor API health and connectivity

• ModelRegistry: Search, compare, and manage deployed models

PredictionService

Service for model prediction and inference operations.

Initialization

PredictionService(http_client, api_client)

Parameters:

• http_client: Configured HTTP client for API requests

• api_client: Legacy API client for backward compatibility

Note: Typically accessed via client.prediction_service property.

execute_prediction(data, model_id, **options)

Execute a prediction (automatically routes to single or batch).

Parameters:

• data (dict or list): Data to predict on

  • dict: Single data point for immediate prediction

  • list: Multiple data points for batch prediction

• model_id (str): Model ID to use for prediction

• **options: Additional prediction options (reserved for future use)

Returns:

• Prediction object for single predictions

• BatchPredictionJob for batch predictions

• None if prediction fails

Raises:

• ValueError: If data format is not supported or model_id is missing

Examples:

# Single prediction
result = service.execute_prediction(
    data={"age": 35, "income": 50000},
    model_id="model-123"
)
print(f"Prediction: {result.prediction}")

# Batch prediction
job = service.execute_prediction(
    data=[{"age": 35}, {"age": 42}],
    model_id="model-123"
)
print(f"Job ID: {job.job_id}")

execute_single_point_prediction(data, model_id)

Execute prediction for a single data point.

Parameters:

• data (dict): Single data point as dictionary of feature values

• model_id (str): Model ID to use for prediction

Returns: Prediction object or None

Raises:

• ValueError: If data is empty or model_id not provided

Example:

result = service.execute_single_point_prediction(
    data={"age": 35, "income": 50000, "credit_score": 720},
    model_id="model-123"
)

if result:
    print(f"Prediction: {result.prediction}")
    print(f"Confidence: {result.confidence}")

Note:

• Data is validated using DataPoint class

• Results returned immediately (synchronous)

• Failed predictions return None with error logged

execute_batch_prediction(data, model_id, wait_for_completion=True)

Execute prediction for multiple data points (batch).

Parameters:

• data (list or DataFrame): List of dictionaries or DataFrame with multiple data points

• model_id (str): Model ID to use for predictions

• wait_for_completion (bool): If True, polls until job completes. If False, returns job immediately.

Returns: BatchPredictionJob object or None

Raises:

• ValueError: If data is empty or model_id not provided

Example:

# Execute and wait for completion
job = service.execute_batch_prediction(
    data=[
        {"age": 35, "income": 50000},
        {"age": 42, "income": 75000},
        {"age": 28, "income": 45000}
    ],
    model_id="model-123",
    wait_for_completion=True
)

if job and job.status == "completed":
    print(f"Results at: {job.dashboard_url}")

# Execute without waiting
job = service.execute_batch_prediction(
    data=data_list,
    model_id="model-123",
    wait_for_completion=False
)

# Monitor job later
status = service.retrieve_batch_job_status(job.job_id)

Note:

• Data is uploaded as CSV to cloud storage

• Job execution is asynchronous

• Results available in dashboard after completion

retrieve_batch_job_status(job_id)

Retrieve the status of a batch prediction job.

Parameters:

• job_id (str): Unique identifier of the batch prediction job

Returns: Dictionary with job status information or None

Example:

status = service.retrieve_batch_job_status("job-123abc")
if status:
    print(f"Status: {status.get('status')}")
    print(f"Progress: {status.get('progress', 0)}%")

monitor_batch_job_until_complete(job, poll_interval=10, max_wait_time=None)

Monitor a batch prediction job until completion.

Parameters:

• job (BatchPredictionJob): BatchPredictionJob object to monitor

• poll_interval (int): Seconds between status checks (default: 10)

• max_wait_time (int, optional): Maximum seconds to wait. If None, uses MAX_POLLING_ITERATIONS * poll_interval

Returns: Updated BatchPredictionJob with final status

Example:

job = service.execute_batch_prediction(
    data=data_list,
    model_id="model-123",
    wait_for_completion=False
)

# Monitor with 5-second intervals, max 5 minutes
completed_job = service.monitor_batch_job_until_complete(
    job=job,
    poll_interval=5,
    max_wait_time=300
)

print(f"Final status: {completed_job.status}")

Note:

• Displays progress bar during monitoring

• Updates job.status in place

• Logs timeout if max wait time exceeded

validate_prediction_data(data, model_id)

Validate data against model schema before prediction.

Parameters:

• data (dict): Data point to validate

• model_id (str): Model ID to validate against

Returns: True if data is valid, False otherwise

Example:

if service.validate_prediction_data(data, model_id):
    result = service.execute_single_point_prediction(data, model_id)
else:
    print("Data validation failed")

ModelTrainingService

Service for model training operations.

Initialization

ModelTrainingService(http_client, api_client)

Note: Typically accessed via client.training_service property.

initiate_training_job(...)

Initiate a new model training job.

Parameters:

• dataset (DataFrame): DataFrame containing training data with target column

• target_column (str): Name of the target/label column in dataset

• workspace (Workspace or str): Workspace object or workspace ID string

• project (Project or str, optional): Optional Project object or project ID string

• model_name (str): Name for the trained model (default: "Default Model")

• description (str): Description of the model/training job (default: "No description provided")

• wait_for_completion (bool): If True, monitors job until completion (default: True)

Returns: TrainingJob object or None

Raises:

• ValueError: If required parameters are missing or invalid

Example:

import pandas as pd

df = pd.read_csv("training_data.csv")

job = service.initiate_training_job(
    dataset=df,
    target_column="churned",
    workspace="ws-123",
    model_name="Customer Churn Model",
    description="Trained on Q4 2024 data",
    wait_for_completion=True
)

if job and job.status == "training_completed":
    print(f"Training complete! View at: {job.dashboard_url}")

Note:

• Training is asynchronous and may take several minutes

• Dataset is uploaded as CSV to cloud storage

• Progress displayed via progress bar if waiting

retrieve_training_job_status(job_id)

Retrieve the current status of a training job.

Parameters:

• job_id (str): Unique identifier of the training job (dashboard_id)

Returns: Dictionary with job status information or None

Example:

status = service.retrieve_training_job_status("job-123abc")
if status:
    print(f"Status: {status.get('status')}")

Note: Status values include: pending, training, training_completed, failed

monitor_training_job_until_complete(job, poll_interval=10, max_wait_time=None)

Monitor a training job until completion or timeout.

Parameters:

• job (TrainingJob): TrainingJob object to monitor

• poll_interval (int): Seconds between status checks (default: 10)

• max_wait_time (int, optional): Maximum seconds to wait

Returns: Updated TrainingJob with final status

Example:

job = service.initiate_training_job(
    dataset=df,
    target_column="target",
    workspace="ws-123",
    wait_for_completion=False
)

# Monitor with 5-second intervals, max 10 minutes
completed_job = service.monitor_training_job_until_complete(
    job=job,
    poll_interval=5,
    max_wait_time=600
)

if completed_job.status == "training_completed":
    print("Training successful!")

Note:

• Displays progress bar during monitoring

• Updates job.status in place

• Prints completion message with dashboard link

retrieve_training_job_results(job_id)

Retrieve results from a completed training job.

Parameters:

• job_id (str): Unique identifier of the completed training job

Returns: Dictionary with training results and metrics or None

Example:

results = service.retrieve_training_job_results("job-123abc")
if results:
    print(f"Model ID: {results.get('model_id')}")
    print(f"Accuracy: {results.get('metrics', {}).get('accuracy')}")

Note:

• Only works for completed training jobs

• Results structure depends on model type

WorkspaceManager

Service for workspace operations.

Initialization

WorkspaceManager(http_client)

Note: Typically accessed via client.workspace_manager property.

retrieve_default_workspace()

Retrieve the default workspace for the authenticated user.

Returns: Workspace object or None

Example:

workspace = manager.retrieve_default_workspace()
if workspace:
    print(f"Default workspace: {workspace.workspace_name}")

Note: Returns the first workspace from the user's accessible workspaces list.

retrieve_workspace_by_name(workspace_name)

Retrieve a workspace by its unique name.

Parameters:

• workspace_name (str): Name of the workspace to retrieve

Returns: Workspace object or None

Raises:

• ValueError: If workspace_name is empty or None

Example:

workspace = manager.retrieve_workspace_by_name("production")
if workspace:
    print(f"Found: {workspace.workspace_id}")
else:
    print("Workspace 'production' not found")

Note: Workspace names are case-sensitive.

retrieve_workspace_by_id(workspace_id)

Retrieve a workspace by its unique identifier.

Parameters:

• workspace_id (str): Unique identifier of the workspace

Returns: Workspace object or None

Raises:

• ValueError: If workspace_id is empty or None

Example:

workspace = manager.retrieve_workspace_by_id("ws-123abc")
if workspace:
    print(f"Workspace name: {workspace.workspace_name}")

list_accessible_workspaces()

List all workspaces accessible to the authenticated user.

Returns: List[Workspace]

Example:

workspaces = manager.list_accessible_workspaces()
print(f"Found {len(workspaces)} workspaces:")

for ws in workspaces:
    print(f"  - {ws.workspace_name} (ID: {ws.workspace_id})")

workspace_exists(workspace_name)

Check if a workspace with the given name exists.

Parameters:

• workspace_name (str): Name of the workspace to check

Returns: True if workspace exists, False otherwise

Example:

if manager.workspace_exists("production"):
    print("Workspace already exists")

count_accessible_workspaces()

Count the number of workspaces accessible to the user.

Returns: Number of accessible workspaces (int)

Example:

count = manager.count_accessible_workspaces()
print(f"You have access to {count} workspaces")

ProjectManager

Service for project operations.

Initialization

ProjectManager(http_client)

Note: Typically accessed via client.project_manager property.

retrieve_project_by_name(project_name)

Retrieve a project by its unique name.

Parameters:

• project_name (str): Name of the project to retrieve

Returns: Project object or None

Raises:

• ValueError: If project_name is empty or None

Example:

project = manager.retrieve_project_by_name("customer-churn")
if project:
    print(f"Found: {project.project_id}")

Note: Project names are case-sensitive.

retrieve_project_by_id(project_id)

Retrieve a project by its unique identifier.

Parameters:

• project_id (str): Unique identifier of the project

Returns: Project object or None

Example:

project = manager.retrieve_project_by_id("proj-123abc")
if project:
    print(f"Project name: {project.project_name}")

list_accessible_projects(workspace_id=None)

List all projects accessible to the authenticated user.

Parameters:

• workspace_id (str, optional): Optional workspace ID to filter projects

Returns: List[Project]

Examples:

# List all projects
projects = manager.list_accessible_projects()
print(f"Found {len(projects)} projects:")

for proj in projects:
    print(f"  - {proj.project_name} (ID: {proj.project_id})")

# List projects in specific workspace
projects = manager.list_accessible_projects(workspace_id="ws-123")

project_exists(project_name)

Check if a project with the given name exists.

Parameters:

• project_name (str): Name of the project to check

Returns: True if project exists, False otherwise

Example:

if manager.project_exists("ml-project"):
    print("Project already exists")

count_accessible_projects(workspace_id=None)

Count the number of projects accessible to the user.

Parameters:

• workspace_id (str, optional): Optional workspace ID to filter count

Returns: Number of accessible projects (int)

Example:

# Count all projects
count = manager.count_accessible_projects()
print(f"You have access to {count} projects")

# Count projects in workspace
count = manager.count_accessible_projects(workspace_id="ws-123")
print(f"Workspace has {count} projects")

search_projects_by_name_pattern(pattern)

Search for projects matching a name pattern.

Parameters:

• pattern (str): Search pattern (case-insensitive substring match)

Returns: List[Project] of matching projects

Example:

projects = manager.search_projects_by_name_pattern("churn")
for proj in projects:
    print(f"Matched: {proj.project_name}")

HealthChecker

Service for API health monitoring and diagnostics.

Initialization

HealthChecker(http_client)

Note: Typically accessed via client.health property.

check_connection()

Check basic connection to API.

Returns: HealthCheckResult with connection status

Example:

health = checker.check_connection()
if health.healthy:
    print("Connected successfully")
    print(f"Response time: {health.response_time_ms}ms")

check_authentication()

Check API authentication.

Returns: HealthCheckResult with authentication status

Example:

auth = checker.check_authentication()
if auth.healthy:
    print("Authentication successful")

check_api_endpoints()

Check multiple API endpoints.

Returns: Dictionary mapping endpoint names to HealthCheckResult objects

Example:

results = checker.check_api_endpoints()
for name, result in results.items():
    status = "✓" if result.healthy else "✗"
    print(f"{status} {name}: {result.response_time_ms}ms")

run_diagnostics()

Run comprehensive diagnostics.

Returns: Dictionary with all diagnostic results

Example:

diagnostics = checker.run_diagnostics()
for check, result in diagnostics.items():
    status = "✓" if result.healthy else "✗"
    print(f"{status} {check}")

measure_latency(num_requests=5)

Measure API latency.

Parameters:

• num_requests (int): Number of requests to make (default: 5)

Returns: Dictionary with latency statistics (min, max, avg, median)

Example:

latency = checker.measure_latency(num_requests=10)
print(f"Min: {latency['min_ms']}ms")
print(f"Max: {latency['max_ms']}ms")
print(f"Avg: {latency['avg_ms']}ms")
print(f"Median: {latency['median_ms']}ms")

test_network_reliability(num_attempts=10)

Test network reliability with multiple requests.

Parameters:

• num_attempts (int): Number of test requests (default: 10)

Returns: Dictionary with reliability statistics

Example:

reliability = checker.test_network_reliability(num_attempts=20)
print(f"Success rate: {reliability['success_rate_pct']}%")
print(f"Failed: {reliability['failed']}/{reliability['total_attempts']}")

ModelRegistry

Service for listing, searching, and managing deployed models.

Initialization

ModelRegistry(http_client)

Note: Typically accessed via client.registry property.

list_models(workspace_id=None, project_id=None, limit=None)

List deployed models.

Parameters:

• workspace_id (str, optional): Filter by workspace ID

• project_id (str, optional): Filter by project ID

• limit (int, optional): Maximum number of models to return

Returns: List[Dict] of model dictionaries

Example:

models = registry.list_models(workspace_id="ws_123")
for model in models:
    print(f"{model['dashboard_name']}: {model['status']}")

get_model(model_id)

Get model by ID.

Parameters:

• model_id (str): Model/dashboard ID

Returns: Model dictionary or None if not found

Example:

model = registry.get_model("model_123")
if model:
    print(f"Model name: {model['dashboard_name']}")

search_models(name=None, tags=None, created_after=None, created_before=None)

Search models by criteria.

Parameters:

• name (str, optional): Search by model name (partial match)

• tags (List[str], optional): Filter by tags

• created_after (datetime, optional): Filter by creation date (after)

• created_before (datetime, optional): Filter by creation date (before)

Returns: List[Dict] of matching models

Example:

from datetime import datetime, timedelta

# Search by name
models = registry.search_models(name="churn")

# Search by tags and date
cutoff = datetime.now() - timedelta(days=30)
recent_models = registry.search_models(
    tags=["production"],
    created_after=cutoff
)

get_model_metadata(model_id)

Get model metadata.

Parameters:

• model_id (str): Model/dashboard ID

Returns: Dictionary with model metadata

Example:

metadata = registry.get_model_metadata("model_123")
print(f"Name: {metadata['dashboard_name']}")
print(f"Type: {metadata['type']}")
print(f"Created: {metadata['created_at']}")

compare_models(model_ids)

Compare multiple models.

Parameters:

• model_ids (List[str]): List of model IDs to compare

Returns: DataFrame comparing models

Example:

comparison = registry.compare_models(["model_1", "model_2", "model_3"])
print(comparison[['dashboard_name', 'type', 'created_at']])

get_model_stats(workspace_id=None)

Get model statistics.

Parameters:

• workspace_id (str, optional): Filter by workspace ID

Returns: Dictionary with model statistics

Example:

stats = registry.get_model_stats()
print(f"Total models: {stats['total_models']}")
print(f"By status: {stats['by_status']}")
print(f"By type: {stats['by_type']}")

get_deployment_history(workspace_id=None, days=30)

Get deployment history.

Parameters:

• workspace_id (str, optional): Filter by workspace ID

• days (int): Number of days of history (default: 30)

Returns: DataFrame with deployment history

Example:

history = registry.get_deployment_history(days=7)
print(history[["dashboard_name", "created_at", "status"]])

find_models_by_name(name_pattern, case_sensitive=False)

Find models by name pattern (partial match).

Parameters:

• name_pattern (str): Pattern to search for in model names

• case_sensitive (bool): Whether search should be case sensitive (default: False)

Returns: List[Dict] of matching models

Example:

churn_models = registry.find_models_by_name("churn")
print(f"Found {len(churn_models)} churn models")

find_recent_models(days=7)

Find models created in the last N days.

Parameters:

• days (int): Number of days to look back (default: 7)

Returns: List[Dict] of recent models

Example:

recent = registry.find_recent_models(days=30)
print(f"Found {len(recent)} models created in last 30 days")

BatchPredictionJob

Represents a batch prediction job.

Attributes

• job_id (str): Unique identifier for the batch prediction job

• model_id (str): Model ID used for predictions

• status (str): Current status of the job

• dashboard_url (str): URL to view job results

Example:

job = BatchPredictionJob(
    job_id="job-123",
    model_id="model-456",
    status="pending",
    dashboard_url="https://app.viewai.ca/dashboard/model-456"
)

print(f"Job: {job.job_id}")
print(f"Status: {job.status}")

TrainingJob

Represents a model training job.

Attributes

• job_id (str): Unique identifier for the training job

• model_name (str): Name of the model being trained

• workspace_id (str): Workspace where model is being trained

• status (str): Current status of the training job

• dashboard_url (str): URL to view training progress/results

Example:

job = TrainingJob(
    job_id="job-789",
    model_name="Customer Churn Model",
    workspace_id="ws-123",
    status="training",
    dashboard_url="https://app.viewai.ca/dashboard/job-789"
)

print(f"Training: {job.model_name}")
print(f"Status: {job.status}")

See Also

• ViewAIClient - Main client class

• Model Classes - Model deployment and prediction classes

• Configuration - Configuration options

• Exceptions - Exception handling