Case 2 — Deploy a Model as a Containerized FastAPI Service

# Case 2 — Deploy a Model as a Containerized FastAPI Service > **Enterprise Scenario:** The churn prediction model registered in Case 1 (`models:/churn_predictor/Production`) is approved for production. Your engineering team needs an inference microservice capable of serving real-time predictions to customer success dashboards. The service must achieve p99 latency < 50ms, load weights into memory at startup (no per-request disk reads), validate feature inputs defensively (e.g. rejection of negative tenures), log telemetry asynchronously without blocking client responses, and package into a secure, minimal Docker container. --- ## Technical Requirements & Deliverables ### Deliverable 1: FastAPI Application (`app/main.py`) - Implement a lifespan context manager (`@asynccontextmanager`) that loads the model into memory at startup. - Implement `/health` (process liveness) and `/ready` (model in memory & dependency check) endpoints. - Implement `/predict` POST endpoint accepting JSON feature payloads and returning probabilities. ### Deliverable 2: Defensive Pydantic Contracts (`app/schemas.py`) - Define `ChurnFeatures` enforcing: - `customer_id`: valid string identifier. - `tenure_months`: integer $\ge 0$ and $\le 120$. - `monthly_charges`: float $> 0.0$ and $\le 1000.0$. - `contract_type`: literal string enum (`"month-to-month"`, `"one-year"`, `"two-year"`). - Define `PredictionResponse` returning `customer_id`, `churn_probability`, `churn_prediction` (0 or 1), `model_version`, and `latency_ms`. - Invalid inputs must return HTTP 422 with structured field error messages. ### Deliverable 3: Asynchronous Ingestion & Telemetry - Leverage FastAPI `BackgroundTasks` to write incoming features and output predictions to an append-only JSONL log file without adding latency to the HTTP response. ### Deliverable 4: Multi-Stage Containerization (`Dockerfile`) - Use a lightweight base image (`python:3.11-slim`). - Utilize multi-stage build layers so build dependencies (gcc, build-essential) are stripped from the final runtime container. - Run the webserver as an unprivileged user (`USER appuser`) for security compliance. - Configure a functional `HEALTHCHECK` directive invoking the `/health` endpoint. --- ## Starter Code Scaffold ```python # app/main.py from contextlib import asynccontextmanager import json, time, mlflow.pyfunc from fastapi import FastAPI, BackgroundTasks, HTTPException, status from pydantic import BaseModel, Field from typing import Literal class ChurnFeatures(BaseModel): customer_id: str tenure_months: int = Field(ge=0, le=120) monthly_charges: float = Field(gt=0.0, le=1000.0) contract_type: Literal["month-to-month", "one-year", "two-year"] class PredictionResponse(BaseModel): customer_id: str churn_probability: float churn_prediction: int model_version: str latency_ms: float ml_models = {} @asynccontextmanager async def lifespan(app: FastAPI): # Load model once into memory ml_models["churn_model"] = mlflow.pyfunc.load_model("models:/churn_predictor/Production") yield ml_models.clear() app = FastAPI(title="Customer Churn Prediction Service", lifespan=lifespan) def log_prediction_background(payload: dict, result: dict, latency_ms: float): # TODO: Append telemetry to JSONL log file pass @app.get("/health") def health(): return {"status": "healthy"} @app.get("/ready") def ready(): if "churn_model" not in ml_models: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Model not loaded") return {"status": "ready"} @app.post("/predict", response_model=PredictionResponse) def predict(features: ChurnFeatures, background_tasks: BackgroundTasks): # TODO: Measure latency, format input vector, invoke model, dispatch background task, return payload pass ``` --- ## 100-Point Graded Rubric * **Criteria 1: Lifespan Caching & Architecture (25 pts)** - *25 pts:* Implements `@asynccontextmanager` loading model once at startup; exposes decoupled `/health` and `/ready` probes; zero per-request disk reads. - *15 pts:* Loads model globally without lifespan context; missing `/ready` check. - *0 pts:* Deserializes model from disk inside the `/predict` request loop. * **Criteria 2: Schema Defense & Boundary Contracts (25 pts)** - *25 pts:* Pydantic models strictly enforce numerical ranges and enum literals; invalid payloads cleanly return HTTP 422. - *15 pts:* Basic type validation present but lacks range boundaries or enum enforcement. - *0 pts:* Accepts unstructured dictionaries without schema validation. * **Criteria 3: Non-Blocking Telemetry & Latency (25 pts)** - *25 pts:* Dispatches prediction logging to `BackgroundTasks`; returns client payloads under 20ms. - *15 pts:* Logs synchronously to disk, adding measurable tail latency to responses. - *0 pts:* Fails to log prediction inputs and outputs. * **Criteria 4: Docker Hardening & Multi-Stage Optimization (25 pts)** -