Deploy Versioned Industrial ML Models as Microservices with Flyte and KServe

"""
Deploying versioned ML models as microservices using Flyte and KServe.
This application ensures secure, scalable, and efficient operations for model inference.
"""
from typing import Dict, Any, List
import os
import logging
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from contextlib import asynccontextmanager

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

class Config:
    """
    Configuration class to manage environment variables.
    """
    flyte_project: str = os.getenv('FLYTE_PROJECT')
    flyte_domain: str = os.getenv('FLYTE_DOMAIN')
    kserve_url: str = os.getenv('KSERVE_URL')

class ModelInput(BaseModel):
    """
    Model input schema to define request structure.
    """
    features: List[float]
    model_version: str

    @validator('features')
    def validate_features(cls, v):
        if not isinstance(v, list) or len(v) == 0:
            raise ValueError('Features must be a non-empty list of floats.')
        return v

app = FastAPI()
config = Config()

@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    Context manager to handle app lifespan tasks.
    """
    logger.info("Starting application...")
    yield
    logger.info("Shutting down application...")

app.add_event_handler("startup", lifespan)

async def fetch_model_prediction(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Fetch model prediction from KServe.
    
    Args:
        data: Input data for prediction.
    Returns:
        JSON response from KServe.
    Raises:
        HTTPException if prediction fails.
    """
    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(f"{config.kserve_url}/predict", json=data)
            response.raise_for_status()  # Raise HTTP error for bad responses
            return response.json()
    except httpx.HTTPStatusError as e:
        logger.error(f"Prediction request failed: {e.response.text}")
        raise HTTPException(status_code=e.response.status_code, detail="Prediction service error")

async def process_and_predict(model_input: ModelInput) -> Dict[str, Any]:
    """
    Process input data and call the prediction service.
    
    Args:
        model_input: Input data for model.
    Returns:
        Prediction result.
    Raises:
        ValueError if input is invalid.
    """
    # Validate input data
    if not model_input:
        raise ValueError('Invalid model input data.')  # Input validation
    # Prepare request data
    request_data = {
        "inputs": model_input.features,
        "model_version": model_input.model_version
    }
    return await fetch_model_prediction(request_data)

@app.post("/predict", response_model=Dict[str, Any])
async def predict(model_input: ModelInput):
    """
    Endpoint for model predictions.
    
    Args:
        model_input: Model input data.
    Returns:
        Prediction response from the model.
    Raises:
        HTTPException if prediction fails.
    """
    try:
        result = await process_and_predict(model_input)
        return result
    except Exception as e:
        logger.error(f"Error processing prediction: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)