Add FastAPI application

app/main.py

@@ -0,0 +1,49 @@
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from app.routers import health, predict, train
+from mlpipeline.exception import MLPipelineException
+import uvicorn
+
+app = FastAPI(
+    title="AutoML MLOps API",
+    description="AutoML pipeline API for heart disease prediction",
+    version="1.0.0"
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(health.router)
+app.include_router(predict.router)
+app.include_router(train.router)
+
+
+@app.exception_handler(MLPipelineException)
+async def mlpipeline_exception_handler(request: Request, exc: MLPipelineException):
+    return JSONResponse(
+        status_code=500,
+        content={"error": str(exc)}
+    )
+
+
+@app.get("/")
+async def root():
+    return {
+        "message": "AutoML MLOps API",
+        "version": "1.0.0",
+        "endpoints": {
+            "health": "/health",
+            "predict": "/predict",
+            "train": "/train"
+        }
+    }
+
+
+if __name__ == "__main__":
+    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

app/routers/health.py

@@ -0,0 +1,31 @@
+from fastapi import APIRouter
+from app.schemas.response import HealthResponse
+from app.utils.model_loader import model_loader
+
+router = APIRouter(prefix="/health", tags=["health"])
+
+
+@router.get("/", response_model=HealthResponse)
+async def health_check():
+    is_loaded = model_loader.is_loaded()
+    return HealthResponse(
+        status="healthy" if is_loaded else "degraded",
+        model_loaded=is_loaded,
+        message="Model loaded and ready" if is_loaded else "Model not loaded"
+    )
+
+
+@router.get("/ready", response_model=HealthResponse)
+async def readiness_check():
+    is_loaded = model_loader.is_loaded()
+    if not is_loaded:
+        return HealthResponse(
+            status="not_ready",
+            model_loaded=False,
+            message="Model not loaded"
+        )
+    return HealthResponse(
+        status="ready",
+        model_loaded=True,
+        message="Service ready"
+    )

app/routers/predict.py

@@ -0,0 +1,61 @@
+from fastapi import APIRouter, HTTPException
+from app.schemas.request import PredictionRequest, BatchPredictionRequest
+from app.schemas.response import PredictionResponse, BatchPredictionResponse
+from app.utils.model_loader import model_loader
+import pandas as pd
+
+router = APIRouter(prefix="/predict", tags=["prediction"])
+
+
+def convert_to_original_columns(data_dict):
+    mapping = {
+        "Chest_pain_type": "Chest pain type",
+        "FBS_over_120": "FBS over 120",
+        "EKG_results": "EKG results",
+        "Max_HR": "Max HR",
+        "Exercise_angina": "Exercise angina",
+        "ST_depression": "ST depression",
+        "Slope_of_ST": "Slope of ST",
+        "Number_of_vessels_fluro": "Number of vessels fluro"
+    }
+    return {mapping.get(k, k): v for k, v in data_dict.items()}
+
+
+def add_interaction_features(df):
+    df['id_x_Age'] = df['id'] * df['Age']
+    return df
+
+
+@router.post("/", response_model=PredictionResponse)
+async def predict_single(request: PredictionRequest):
+    try:
+        pipeline = model_loader.get_pipeline()
+        input_dict = convert_to_original_columns(request.model_dump())
+        df = pd.DataFrame([input_dict])
+        df = add_interaction_features(df)
+        result = pipeline.predict(df)
+        
+        return PredictionResponse(
+            prediction=result["predictions"][0],
+            probability=result.get("probabilities")[0] if result.get("probabilities") else None
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.post("/batch", response_model=BatchPredictionResponse)
+async def predict_batch(request: BatchPredictionRequest):
+    try:
+        pipeline = model_loader.get_pipeline()
+        data_list = [convert_to_original_columns(item.model_dump()) for item in request.data]
+        df = pd.DataFrame(data_list)
+        df = add_interaction_features(df)
+        result = pipeline.predict(df)
+        
+        return BatchPredictionResponse(
+            predictions=result["predictions"],
+            probabilities=result.get("probabilities"),
+            num_samples=result["num_samples"]
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

app/routers/train.py

@@ -0,0 +1,49 @@
+from fastapi import APIRouter, HTTPException, BackgroundTasks
+from app.schemas.request import TrainingRequest
+from app.schemas.response import TrainingResponse
+from app.utils.model_loader import model_loader
+from mlpipeline.pipeline.training_pipeline import TrainingPipeline
+
+router = APIRouter(prefix="/train", tags=["training"])
+
+
+def run_training_task():
+    try:
+        pipeline = TrainingPipeline()
+        artifact = pipeline.run_pipeline()
+        model_loader.reload_model()
+        return artifact
+    except Exception as e:
+        raise e
+
+
+@router.post("/", response_model=TrainingResponse)
+async def trigger_training(request: TrainingRequest, background_tasks: BackgroundTasks):
+    try:
+        if request.force_retrain:
+            background_tasks.add_task(run_training_task)
+            return TrainingResponse(
+                status="training_started",
+                message="Training pipeline started in background"
+            )
+        else:
+            artifact = run_training_task()
+            return TrainingResponse(
+                status="completed",
+                message="Training completed successfully",
+                model_path=artifact.pushed_model_path
+            )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.post("/reload", response_model=TrainingResponse)
+async def reload_model():
+    try:
+        model_loader.reload_model()
+        return TrainingResponse(
+            status="success",
+            message="Model reloaded successfully"
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

app/schemas/request.py

@@ -0,0 +1,30 @@
+from pydantic import BaseModel, Field
+from typing import List, Optional
+
+
+class PredictionRequest(BaseModel):
+    id: int
+    Age: int
+    Sex: int
+    Chest_pain_type: int = Field(alias="Chest pain type")
+    BP: int
+    Cholesterol: int
+    FBS_over_120: int = Field(alias="FBS over 120")
+    EKG_results: int = Field(alias="EKG results")
+    Max_HR: int = Field(alias="Max HR")
+    Exercise_angina: int = Field(alias="Exercise angina")
+    ST_depression: float = Field(alias="ST depression")
+    Slope_of_ST: int = Field(alias="Slope of ST")
+    Number_of_vessels_fluro: int = Field(alias="Number of vessels fluro")
+    Thallium: int
+
+    class Config:
+        populate_by_name = True
+
+
+class BatchPredictionRequest(BaseModel):
+    data: List[PredictionRequest]
+
+
+class TrainingRequest(BaseModel):
+    force_retrain: Optional[bool] = False

app/schemas/response.py

@@ -0,0 +1,26 @@
+from pydantic import BaseModel
+from typing import List, Optional, Dict, Any
+
+
+class PredictionResponse(BaseModel):
+    prediction: int
+    probability: Optional[List[float]] = None
+
+
+class BatchPredictionResponse(BaseModel):
+    predictions: List[int]
+    probabilities: Optional[List[List[float]]] = None
+    num_samples: int
+
+
+class HealthResponse(BaseModel):
+    status: str
+    model_loaded: bool
+    message: str
+
+
+class TrainingResponse(BaseModel):
+    status: str
+    message: str
+    metrics: Optional[Dict[str, Any]] = None
+    model_path: Optional[str] = None

app/utils/model_loader.py

@@ -0,0 +1,32 @@
+from mlpipeline.pipeline.prediction_pipeline import PredictionPipeline
+from threading import Lock
+
+
+class ModelLoader:
+    _instance = None
+    _lock = Lock()
+    
+    def __new__(cls):
+        if cls._instance is None:
+            with cls._lock:
+                if cls._instance is None:
+                    cls._instance = super().__new__(cls)
+                    cls._instance.pipeline = None
+        return cls._instance
+    
+    def get_pipeline(self) -> PredictionPipeline:
+        if self.pipeline is None:
+            self.pipeline = PredictionPipeline()
+            self.pipeline.load_model()
+        return self.pipeline
+    
+    def reload_model(self):
+        self.pipeline = PredictionPipeline()
+        self.pipeline.load_model()
+        return self.pipeline
+    
+    def is_loaded(self) -> bool:
+        return self.pipeline is not None and self.pipeline.model is not None
+
+
+model_loader = ModelLoader()

test_api.py

@@ -0,0 +1,85 @@
+import requests
+import json
+
+BASE_URL = "http://localhost:8000"
+
+
+def test_health():
+    response = requests.get(f"{BASE_URL}/health/")
+    print(f"Health Check: {response.json()}")
+
+
+def test_predict():
+    data = {
+        "id": 1,
+        "Age": 50,
+        "Sex": 1,
+        "Chest pain type": 2,
+        "BP": 120,
+        "Cholesterol": 200,
+        "FBS over 120": 0,
+        "EKG results": 0,
+        "Max HR": 150,
+        "Exercise angina": 0,
+        "ST depression": 1.0,
+        "Slope of ST": 1,
+        "Number of vessels fluro": 0,
+        "Thallium": 3
+    }
+    response = requests.post(f"{BASE_URL}/predict/", json=data)
+    print(f"Prediction: {response.json()}")
+
+
+def test_batch_predict():
+    data = {
+        "data": [
+            {
+                "id": 1,
+                "Age": 50,
+                "Sex": 1,
+                "Chest pain type": 2,
+                "BP": 120,
+                "Cholesterol": 200,
+                "FBS over 120": 0,
+                "EKG results": 0,
+                "Max HR": 150,
+                "Exercise angina": 0,
+                "ST depression": 1.0,
+                "Slope of ST": 1,
+                "Number of vessels fluro": 0,
+                "Thallium": 3
+            },
+            {
+                "id": 2,
+                "Age": 60,
+                "Sex": 0,
+                "Chest pain type": 1,
+                "BP": 130,
+                "Cholesterol": 220,
+                "FBS over 120": 1,
+                "EKG results": 1,
+                "Max HR": 140,
+                "Exercise angina": 1,
+                "ST depression": 2.0,
+                "Slope of ST": 2,
+                "Number of vessels fluro": 1,
+                "Thallium": 6
+            }
+        ]
+    }
+    response = requests.post(f"{BASE_URL}/predict/batch", json=data)
+    print(f"Batch Prediction: {response.json()}")
+
+
+if __name__ == "__main__":
+    print("Testing API Endpoints...")
+    print("-" * 50)
+    
+    try:
+        test_health()
+        print("-" * 50)
+        test_predict()
+        print("-" * 50)
+        test_batch_predict()
+    except Exception as e:
+        print(f"Error: {e}")

test_train_endpoints.py

@@ -0,0 +1,28 @@
+import requests
+
+BASE_URL = "http://localhost:8000"
+
+
+def test_reload():
+    print("Testing model reload...")
+    response = requests.post(f"{BASE_URL}/train/reload")
+    print(f"Response: {response.json()}\n")
+
+
+def test_health_after_load():
+    print("Testing health after model load...")
+    response = requests.get(f"{BASE_URL}/health/")
+    print(f"Response: {response.json()}\n")
+
+
+def test_ready():
+    print("Testing readiness endpoint...")
+    response = requests.get(f"{BASE_URL}/health/ready")
+    print(f"Response: {response.json()}\n")
+
+
+if __name__ == "__main__":
+    test_reload()
+    test_health_after_load()
+    test_ready()
+    print("All endpoint tests completed!")