Case 2 — Deploy a Model as a Containerized FastAPI Service

# Case 2 — Deploy a Model as a Containerized FastAPI Service > *"Your model is in MLflow's registry as `churn_predictor:Production`. Build a FastAPI service that loads it, exposes a `/predict` endpoint, packages as a Docker image, and deploys to Cloud Run (or equivalent)."* --- ## Step 1 — Project structure ``` churn-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI app │ ├── models.py # Pydantic models │ └── features.py # Imported from training: same build_features ├── tests/ │ └── test_predict.py ├── Dockerfile ├── requirements.txt ├── .dockerignore └── README.md ``` ## Step 2 — Pydantic schemas `app/models.py`: ```python from pydantic import BaseModel, Field from datetime import datetime from typing import Optional class PredictionRequest(BaseModel): customer_id: str = Field(..., min_length=1, max_length=64) tenure_months: int = Field(..., ge=0, le=600) monthly_revenue: float = Field(..., ge=0) plan: str = Field(..., min_length=1) signup_channel: str country: str logins_last_30d: int = Field(..., ge=0) feature_usage_score: float = Field(..., ge=0, le=100) support_tickets_last_30d: int = Field(..., ge=0) class PredictionResponse(BaseModel): customer_id: str churn_probability: float model_version: str served_at: datetime ``` ## Step 3 — FastAPI service `app/main.py`: ```python import logging from datetime import datetime import mlflow.sklearn from fastapi import FastAPI, HTTPException, BackgroundTasks from .models import PredictionRequest, PredictionResponse from .features import build_features logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) MODEL_NAME = 'churn_predictor' MODEL_STAGE = 'Production' BASELINE_PROB = 0.05 # fallback if model fails app = FastAPI(title='Churn Prediction Service') model = None model_version = None @app.on_event('startup') def load_model(): global model, model_version try: client = mlflow.tracking.MlflowClient() latest = client.get_latest_versions(MODEL_NAME, stages=[MODEL_STAGE])[0] model = mlflow.sklearn.load_model(f'models:/{MODEL_NAME}/{MODEL_STAGE}') model_version = latest.version logger.info(f'Loaded {MODEL_NAME} v{model_version}') except Exception as e: logger.exception(f'Failed to load model: {e}') raise @app.get('/health') def health(): return {'status': 'ok'} @app.get('/ready') def ready(): if model is None: raise HTTPException(status_code=503, detail='Model not loaded') return {'status': 'ready', 'model_version': model_version} @app.post('/predict', response_model=PredictionResponse) def predict(req: PredictionRequest, background_tasks: BackgroundTasks): try: features = build_features(req.dict()) prob = float(model.predict_proba([features])[0][1]) except Exception: logger.exception('Prediction failed') prob = BASELINE_PROB version = 'fallback' background_tasks.add_task( log_prediction, customer_id=req.customer_id, prob=prob, version=model_version, ) return PredictionResponse( customer_id=req.customer_id, churn_probability=prob, model_version=model_version, served_at=datetime.utcnow(), ) def log_prediction(customer_id, prob, version): # In production: write to data warehouse, S3, etc. logger.info(f'Logged prediction: customer={customer_id} prob={prob:.4f} version={version}') ``` ## Step 4 — Tests `tests/test_predict.py`: ```python from fastapi.testclient import TestClient from app.main import app client = TestClient(app) def test_health(): r = client.get('/health') assert r.status_code == 200 def test_predict_valid(): r = client.post('/predict', json={ 'customer_id': 'test_1', 'tenure_months': 12, 'monthly_revenue': 49.99, 'plan': 'pro', 'signup_channel': 'organic', 'country': 'US', 'logins_last_30d': 15, 'feature_usage_score': 72.5, 'support_tickets_last_30d': 1, }) assert r.status_code == 200 body = r.json() assert 0 <= body['churn_probability'] <= 1 assert body['model_version'] is not None def test_predict_invalid(): r = client.post('/predict', json={ 'customer_id': 'test', 'tenure_months': -5, # invalid 'monthly_revenue': 49.99, 'plan': 'pro', 'signup_channel': 'organic', 'country': 'US', 'logins_last_30d': 15, 'feature_usage_score': 72.5, 'support_tickets_last_30d': 1, }) assert r.status_code == 422 # validation error ``` Run: `pytest`. All tests must pass before deploying. ## Step 5 — Dockerfile ```dockerfile FROM python:3.11.5-slim AS builder WORKDIR /app COPY requirements.txt . RUN pip install --user --no-cache-dir -r requirements.txt FROM python:3.11.5-slim WORKDIR /app # Non-root user RUN