Serve and Store Industrial AI Model Artifacts from On-Premises Object Storage with MinIO and BentoML

"""
Production implementation for serving and storing industrial AI model artifacts using MinIO and BentoML.
This code provides secure and scalable operations for model artifact management.
"""

from typing import Dict, Any, List, Tuple
import os
import logging
import time
import boto3
from botocore.exceptions import ClientError
from pydantic import BaseModel, ValidationError

# Logger setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """Configuration class for environment variables."""
    minio_endpoint: str = os.getenv('MINIO_ENDPOINT')
    access_key: str = os.getenv('MINIO_ACCESS_KEY')
    secret_key: str = os.getenv('MINIO_SECRET_KEY')
    bucket_name: str = os.getenv('MINIO_BUCKET_NAME')

class ModelArtifact(BaseModel):
    """Pydantic model for validating model artifacts."""
    model_id: str
    version: str
    metadata: Dict[str, Any]

def validate_input(data: Dict[str, Any]) -> ModelArtifact:
    """Validate request data for model artifact.
    
    Args:
        data: Input to validate
    Returns:
        ModelArtifact: Validated model artifact
    Raises:
        ValueError: If validation fails
    """
    try:
        return ModelArtifact(**data)
    except ValidationError as e:
        logger.error(f'Validation error: {e}')
        raise ValueError('Invalid input data')

def sanitize_fields(data: ModelArtifact) -> ModelArtifact:
    """Sanitize the fields of the model artifact.
    
    Args:
        data: ModelArtifact to sanitize
    Returns:
        ModelArtifact: Sanitized model artifact
    """
    # Example of sanitization: trimming string fields
    data.model_id = data.model_id.strip()
    return data

def create_minio_client() -> boto3.client:
    """Create a MinIO client.
    
    Returns:
        boto3.client: MinIO client
    """
    return boto3.client(
        's3',
        endpoint_url=Config.minio_endpoint,
        aws_access_key_id=Config.access_key,
        aws_secret_access_key=Config.secret_key,
        region_name='us-east-1'
    )

def upload_to_minio(client: boto3.client, bucket_name: str, file_path: str, object_name: str) -> None:
    """Upload a file to MinIO.
    
    Args:
        client: MinIO client
        bucket_name: Name of the bucket
        file_path: Path of the file to upload
        object_name: Name of the object in the bucket
    Raises:
        Exception: If upload fails
    """
    try:
        client.upload_file(file_path, object_name)
        logger.info(f'Successfully uploaded {file_path} to {bucket_name}/{object_name}')
    except ClientError as e:
        logger.error(f'Failed to upload {file_path}: {e}')
        raise Exception('Upload failed')

def fetch_data() -> List[Dict[str, Any]]:
    """Fetch data to process. This function simulates data fetching.
    
    Returns:
        List[Dict[str, Any]]: Sample data
    """
    return [
        {'model_id': 'model_1', 'version': 'v1.0', 'metadata': {'description': 'First model'}},
        {'model_id': 'model_2', 'version': 'v2.0', 'metadata': {'description': 'Second model'}}
    ]

def process_batch(data: List[Dict[str, Any]]) -> List[Tuple[str, str]]:
    """Process a batch of model artifacts.
    
    Args:
        data: List of model artifacts to process
    Returns:
        List[Tuple[str, str]]: List of processed artifacts
    """
    processed = []
    for item in data:
        validated_item = validate_input(item)  # Validate each item
        sanitized_item = sanitize_fields(validated_item)  # Sanitize
        processed.append((sanitized_item.model_id, sanitized_item.version))
    return processed

def save_to_db(artifacts: List[Tuple[str, str]]) -> None:
    """Simulate saving processed artifacts to a database.
    
    Args:
        artifacts: List of processed artifacts
    """
    # Simulate database save with a simple print
    for artifact in artifacts:
        logger.info(f'Saving artifact {artifact[0]} version {artifact[1]} to database')

def handle_errors(func):
    """Decorator to handle errors gracefully.
    
    Args:
        func: Function to decorate
    Returns:
        Callable: Wrapped function
    """
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f'Error in {func.__name__}: {e}')
            return None
    return wrapper

@handle_errors
def main() -> None:
    """Main function to orchestrate the model artifact workflow.
    
    Returns:
        None
    """
    client = create_minio_client()  # Create MinIO client
    data = fetch_data()  # Fetch data
    processed_artifacts = process_batch(data)  # Process batch
    save_to_db(processed_artifacts)  # Save to DB
    for artifact in processed_artifacts:
        # Upload each artifact to MinIO
        upload_to_minio(client, Config.bucket_name, f'{artifact[0]}.zip', f'models/{artifact[0]}/{artifact[1]}.zip')

if __name__ == '__main__':
    main()  # Example usage of the workflow