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

# 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)
  • 25 pts: Multi-stage build under 250MB; runs under unprivileged non-root user; includes functional HEALTHCHECK.
  • 15 pts: Standard single-stage image running as root; image size exceeds 800MB.
  • 0 pts: Dockerfile fails to build or lacks dependency installation caching.