Serving with FastAPI

from fastapi import FastAPI
from pydantic import BaseModel
import joblib
 
app = FastAPI()
model = joblib.load("model.joblib")
 
class PredictRequest(BaseModel):
    features: list[float]
 
@app.post("/predict")
def predict(req: PredictRequest) -> dict:
    pred = model.predict([req.features])
    return {"prediction": pred.tolist()}

"""serving_fastapi.py - production-ready model API."""
from __future__ import annotations
 
from contextlib import asynccontextmanager
from typing import Any
 
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
 
MODEL_PATH = "models/churn_pipeline.joblib"
MODEL_VERSION = "1.2.0"
model: Any = None
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    global model
    model = joblib.load(MODEL_PATH)
    yield
    model = None
 
app = FastAPI(title="Churn Predictor", version=MODEL_VERSION, lifespan=lifespan)
 
class PredictRequest(BaseModel):
    tenure: float = Field(ge=0)
    monthly_charges: float = Field(ge=0)
    total_charges: float = Field(ge=0)
 
class PredictResponse(BaseModel):
    churn_probability: float
    churn_prediction: bool
    model_version: str
 
@app.get("/health")
def health() -> dict:
    return {"status": "ok", "model_loaded": model is not None}
 
@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest) -> PredictResponse:
    if model is None:
        raise HTTPException(503, "Model not loaded")
    features = np.array([[req.tenure, req.monthly_charges, req.total_charges]])
    proba = model.predict_proba(features)[0][1]
    return PredictResponse(
        churn_probability=round(float(proba), 4),
        churn_prediction=proba >= 0.5,
        model_version=MODEL_VERSION,
    )