Best Practices

Production-ready patterns and best practices for the ViewAI Python SDK.

Security

API Key Management

❌ Never commit API keys to version control

# DON'T DO THIS
client = ViewAIClient(api_key="835fa411ae67ed02...")  # Hardcoded key

✅ Use environment variables

import os

# Load from environment
api_key = os.getenv("VIEWAI_API_KEY")
if not api_key:
    raise ValueError("VIEWAI_API_KEY environment variable not set")

client = ViewAIClient(api_key=api_key)

✅ Use configuration files (outside version control)

import json

# Load from config file (add to .gitignore)
with open("config.json") as f:
    config = json.load(f)

client = ViewAIClient(api_key=config["api_key"])

✅ Use secret management services

# AWS Secrets Manager
import boto3
import json

def get_secret(secret_name):
    client = boto3.client("secretsmanager", region_name="us-east-1")
    response = client.get_secret_value(SecretId=secret_name)
    return json.loads(response["SecretString"])

secrets = get_secret("viewai/api-credentials")
client = ViewAIClient(api_key=secrets["api_key"])

Secure Logging

import logging

# Don't log sensitive data
logger = logging.getLogger(__name__)

# ❌ Don't do this
logger.info(f"Using API key: {api_key}")

# ✅ Do this
logger.info("ViewAI client initialized")

# ✅ Mask sensitive data if necessary
masked_key = api_key[:8] + "..." if api_key else "None"
logger.debug(f"API key starts with: {masked_key}")

Error Handling

Always Use Try-Except

from viewai_client.exceptions import ViewAIError, APIError, NetworkError

try:
    prediction = client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )

    if prediction:
        result = prediction.get_raw_prediction()
        # Process result
    else:
        logger.warning("Prediction returned None")

except APIError as e:
    logger.error(f"API error: {e.message} (status: {e.status_code})")
    # Handle API error (maybe retry)

except NetworkError as e:
    logger.error(f"Network error: {e.message}")
    # Handle network error

except ViewAIError as e:
    logger.error(f"ViewAI error: {e.message}")
    # Handle general ViewAI error

except Exception as e:
    logger.error(f"Unexpected error: {e}")
    # Handle unexpected error

Implement Retry Logic

from tenacity import retry, stop_after_attempt, wait_exponential
from viewai_client.exceptions import NetworkError, APIError

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=lambda e: isinstance(e, (NetworkError, APIError))
)
def make_prediction_with_retry(client, data, model_id):
    """Make prediction with automatic retry"""
    return client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )

# Usage
prediction = make_prediction_with_retry(client, data, model_id)

Configuration Management

Use Configuration Objects

from viewai_client import ViewAIClient, ClientConfiguration

# ❌ Don't pass many parameters
client = ViewAIClient(
    api_key=api_key,
    timeout=30,
    max_retries=3,
    # ... many more parameters
)

# ✅ Use configuration object
config = ClientConfiguration(
    api_key=api_key,
    timeout=30,
    max_retries=3,
    connection_timeout=10,
    read_timeout=30,
    log_level=logging.INFO
)

client = ViewAIClient(config=config)

Environment-Specific Configuration

import os

def get_config_for_environment():
    """Get configuration based on environment"""

    env = os.getenv("ENVIRONMENT", "development")

    if env == "production":
        return ClientConfiguration(
            api_key=os.getenv("VIEWAI_API_KEY"),
            timeout=60,
            max_retries=5,
            log_level=logging.INFO
        )
    elif env == "staging":
        return ClientConfiguration(
            api_key=os.getenv("VIEWAI_API_KEY_STAGING"),
            timeout=30,
            max_retries=3,
            log_level=logging.DEBUG
        )
    else:  # development
        return ClientConfiguration(
            api_key=os.getenv("VIEWAI_API_KEY_DEV"),
            timeout=30,
            max_retries=2,
            log_level=logging.DEBUG
        )

# Usage
config = get_config_for_environment()
client = ViewAIClient(config=config)

Resource Management

Use Context Managers

# ✅ Automatic resource cleanup
with ViewAIClient(api_key=api_key) as client:
    prediction = client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )
    # Client resources automatically cleaned up

Manual Cleanup

client = ViewAIClient(api_key=api_key)

try:
    # Use client
    prediction = client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )
finally:
    # Always cleanup
    client.cleanup_resources()

Data Validation

Validate Before Prediction

def validate_input_data(data, required_fields):
    """Validate input data before prediction"""

    # Check required fields
    missing = [f for f in required_fields if f not in data]
    if missing:
        raise ValueError(f"Missing required fields: {missing}")

    # Check data types
    if "age" in data and not isinstance(data["age"], (int, float)):
        raise ValueError("age must be numeric")

    # Check ranges
    if "age" in data and (data["age"] < 0 or data["age"] > 120):
        raise ValueError("age must be between 0 and 120")

    return True

# Usage
required_fields = ["age", "income", "tenure"]

try:
    validate_input_data(data, required_fields)
    prediction = client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )
except ValueError as e:
    logger.error(f"Validation error: {e}")

Schema Validation Before Deployment

# Validate data against schema before deploying
model_obj = client.create_deployable_model(
    model=trained_model,
    dataset=train_df,
    target_column="target",
    model_name="My Model",
    workspace=workspace
)

# Validate test data
validation_result = model_obj.validate_data(test_df)

if not validation_result["valid"]:
    print("Validation failed!")
    for error in validation_result["errors"]:
        print(f"  - {error}")
    # Fix issues before deploying
else:
    # Deploy only if validation passes
    dashboard_id = model_obj.deploy()

Performance

Use Batch for Multiple Predictions

# ❌ Inefficient
for row in df.iterrows():
    prediction = client.execute_single_point_prediction(
        data=row[1].to_dict(),
        model_id=model_id
    )

# ✅ Efficient
batch_job = client.execute_batch_prediction(
    data=df,
    model_id=model_id,
    wait_for_completion=True
)

Implement Caching

from functools import lru_cache
import hashlib
import json

@lru_cache(maxsize=1000)
def get_cached_prediction(data_hash, model_id):
    """Cached prediction wrapper"""
    # This won't be called if result is cached
    data = json.loads(data_hash)  # Reconstruct data
    return client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )

def predict_with_cache(data, model_id):
    """Make prediction with caching"""
    # Create hashable key
    data_hash = json.dumps(data, sort_keys=True)
    return get_cached_prediction(data_hash, model_id)

Testing

Unit Testing

import unittest
from unittest.mock import Mock, patch

class TestViewAIIntegration(unittest.TestCase):

    def setUp(self):
        self.api_key = "test-api-key"

    @patch("viewai_client.core.client.HTTPClient")
    def test_prediction(self, mock_http):
        """Test prediction with mocked HTTP client"""

        # Setup mock
        mock_http.return_value.post.return_value = {
            "predicted_value": 1,
            "probability_1": 0.8
        }

        # Test
        client = ViewAIClient(api_key=self.api_key)
        prediction = client.execute_single_point_prediction(
            data={"age": 35},
            model_id="test-model"
        )

        # Assertions
        self.assertIsNotNone(prediction)

    def test_configuration_validation(self):
        """Test configuration validation"""

        with self.assertRaises(ValueError):
            # Should raise - no API key
            client = ViewAIClient()

Integration Testing

import pytest
import os

@pytest.fixture
def client():
    """Fixture for ViewAI client"""
    api_key = os.getenv("VIEWAI_TEST_API_KEY")
    if not api_key:
        pytest.skip("Test API key not configured")

    return ViewAIClient(api_key=api_key)

def test_health_check(client):
    """Test API health check"""
    health = client.health.check_connection()
    assert health.get("healthy") is True

def test_single_prediction(client):
    """Test single prediction"""
    data = {"age": 35, "income": 75000}
    model_id = os.getenv("VIEWAI_TEST_MODEL_ID")

    prediction = client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )

    assert prediction is not None
    assert "predicted" in str(prediction).lower()

Logging

Configure Structured Logging

import logging
import json

class StructuredLogger:
    """Structured logging for ViewAI operations"""

    def __init__(self, name):
        self.logger = logging.getLogger(name)

    def log_prediction(self, model_id, data, prediction, elapsed_time):
        """Log prediction with structured data"""
        log_data = {
            "event": "prediction",
            "model_id": model_id,
            "data_size": len(str(data)),
            "elapsed_ms": elapsed_time * 1000,
            "success": prediction is not None
        }
        self.logger.info(json.dumps(log_data))

    def log_error(self, operation, error, context=None):
        """Log error with context"""
        log_data = {
            "event": "error",
            "operation": operation,
            "error_type": type(error).__name__,
            "error_message": str(error),
            "context": context or {}
        }
        self.logger.error(json.dumps(log_data))

# Usage
logger = StructuredLogger(__name__)

import time
start_time = time.time()

try:
    prediction = client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )
    elapsed = time.time() - start_time
    logger.log_prediction(model_id, data, prediction, elapsed)

except Exception as e:
    logger.log_error("prediction", e, {"model_id": model_id})

Production Deployment

Docker Deployment

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Set environment variables
ENV PYTHONUNBUFFERED=1

# Run application
CMD ["python", "app.py"]

# app.py
import os
from viewai_client import ViewAIClient

def main():
    # Get configuration from environment
    api_key = os.getenv("VIEWAI_API_KEY")
    model_id = os.getenv("VIEWAI_MODEL_ID")

    if not api_key:
        raise ValueError("VIEWAI_API_KEY not set")

    # Initialize client
    client = ViewAIClient(api_key=api_key)

    # Your application logic
    print("ViewAI client initialized")

if __name__ == "__main__":
    main()

Kubernetes Deployment

# deployment.yaml
apiVersion: v1
kind: Secret
metadata:
  name: viewai-secrets
type: Opaque
stringData:
  api-key: "your-api-key-here"

---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: viewai-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: viewai-app
  template:
    metadata:
      labels:
        app: viewai-app
    spec:
      containers:
      - name: app
        image: your-registry/viewai-app:latest
        env:
        - name: VIEWAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: viewai-secrets
              key: api-key
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"

Monitoring

Health Checks

def run_health_checks(client):
    """Run comprehensive health checks"""

    checks = {
        "connection": False,
        "authentication": False,
        "api_endpoints": False
    }

    try:
        # Check connection
        conn_result = client.health.check_connection()
        checks["connection"] = conn_result.get("healthy", False)

        # Check authentication
        auth_result = client.health.check_authentication()
        checks["authentication"] = auth_result.get("valid", False)

        # Check API endpoints
        api_result = client.health.check_api_endpoints()
        checks["api_endpoints"] = api_result.get("all_available", False)

    except Exception as e:
        logger.error(f"Health check failed: {e}")

    return checks

# Schedule health checks
import schedule

schedule.every(5).minutes.do(lambda: run_health_checks(client))

Metrics Collection

from prometheus_client import Counter, Histogram
import time

# Define metrics
prediction_counter = Counter(
    "viewai_predictions_total",
    "Total number of predictions",
    ["model_id", "status"]
)

prediction_latency = Histogram(
    "viewai_prediction_latency_seconds",
    "Prediction latency in seconds",
    ["model_id"]
)

def make_monitored_prediction(client, data, model_id):
    """Make prediction with metrics collection"""

    start_time = time.time()
    status = "success"

    try:
        prediction = client.execute_single_point_prediction(
            data=data,
            model_id=model_id
        )

        if prediction is None:
            status = "failed"

        return prediction

    except Exception as e:
        status = "error"
        raise

    finally:
        # Record metrics
        elapsed = time.time() - start_time
        prediction_counter.labels(model_id=model_id, status=status).inc()
        prediction_latency.labels(model_id=model_id).observe(elapsed)