Package and Autoscale Industrial Model APIs with Ray and KServe

"""
Production implementation for packaging and autoscaling industrial model APIs using Ray and KServe.
Provides secure, scalable operations.
"""
from typing import Dict, Any, List
import os
import logging
import requests
from pydantic import BaseModel, ValidationError
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from ray import serve
import time

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

# Configuration class for environment variables
class Config:
    model_endpoint: str = os.getenv('MODEL_ENDPOINT', 'http://localhost:8000/predict')
    max_retries: int = 5
    backoff_factor: float = 0.3

# FastAPI app initialization
app = FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=['*'], allow_credentials=True, allow_methods=['*'], allow_headers=['*'])

# Data model
class InputData(BaseModel):
    features: List[float]  # Input features for the model

class OutputData(BaseModel):
    prediction: float  # Model prediction output

def validate_input(data: Dict[str, Any]) -> None:
    """
    Validate input data for model prediction.
    
    Args:
        data: Input data to validate
    Raises:
        ValueError: If validation fails
    """
    if 'features' not in data:
        raise ValueError('Missing features')
    if not isinstance(data['features'], list):
        raise ValueError('Features must be a list')
    if not all(isinstance(x, (int, float)) for x in data['features']):
        raise ValueError('All features must be numeric')

def fetch_data(url: str, json_data: Dict[str, Any]) -> OutputData:
    """
    Fetch prediction from the model API.
    
    Args:
        url: The model endpoint URL
        json_data: Data to send in the request
    Returns:
        OutputData: Model prediction data
    Raises:
        HTTPException: If prediction fails
    """
    for attempt in range(Config.max_retries):
        try:
            response = requests.post(url, json=json_data)
            response.raise_for_status()
            return OutputData(**response.json())
        except requests.exceptions.RequestException as e:
            logger.warning(f'Error fetching data: {e}')
            time.sleep(Config.backoff_factor * (2 ** attempt))  # Exponential backoff
    raise HTTPException(status_code=500, detail='Failed to fetch prediction from model API')

@app.post("/predict", response_model=OutputData)
async def predict(input_data: InputData) -> JSONResponse:
    """
    Endpoint for model prediction.
    
    Args:
        input_data: InputData containing features
    Returns:
        JSONResponse: Prediction result
    Raises:
        HTTPException: If validation or fetching fails
    """
    try:
        # Validate input data
        validate_input(input_data.dict())
        # Prepare data for API call
        json_data = {'features': input_data.features}
        # Fetch prediction from the model API
        result = fetch_data(Config.model_endpoint, json_data)
        return JSONResponse(content=result.dict())
    except ValueError as e:
        logger.error(f'Input validation error: {e}')
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.error(f'Unexpected error: {e}')
        raise HTTPException(status_code=500, detail='An unexpected error occurred')

if __name__ == '__main__':
    # Example usage
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)  # Run FastAPI server