Sandeep Bazar 365 Days of AI
Labs

Machine Learning › Machine Learning in Practice › Day 194

Hands-on lab — Day 194: Serving a Model over an API

Commands

Setup

pip install -r requirements/requirements.txt

Run

python3 examples/serving_a_model_over_an_api_lib.py

Test

./tests/run_tests.sh

File tree

examples/serving_a_model_over_an_api_lib.py
examples/test_serving_a_model_over_an_api_lib.py
expected-output/examples-run.txt
expected-output/FIELDS.md
expected-output/measured-values.txt
expected-output/starter-run.txt
expected-output/test-run.txt
metadata.yml
README.md
requirements/requirements.txt
security.md
starter/serving_a_model_over_an_api_lib.py
starter/test_serving_a_model_over_an_api_lib.py
tests/run_tests.sh
tests/test_serving_a_model_over_an_api_lib.py
troubleshooting.md

Lab README

Lab: Day 194 -- Serving a Model over an API

Lesson

Day number: 194 of 365. Course: Course04-SS03 (Beyond Supervised Learning). Topic: FastAPI Microservice Serving, Pydantic Schema Contracts, and Circuit Breakers.

Purpose

Build a complete, low-latency REST Model Serving Engine in pure Python and NumPy. You will implement preloaded in-memory model execution, Pydantic-style feature boundary validation, vectorized batch inference, Kubernetes health probes, and automated fallback circuit breakers.

Learning objectives

  • Architect preloaded in-memory model inference services.
  • Enforce schema contracts with strict type and boundary validations.
  • Implement single-item and vectorized batch prediction endpoints.
  • Build Kubernetes-compatible liveness and readiness health probes.
  • Guarantee uptime via automated fallback circuit breakers.

Prerequisites

  • Linear models (logistic sigmoid activation).
  • Python 3.11+ dataclasses and NumPy.

Supported operating systems

  • macOS (Apple Silicon / Intel)
  • Linux (Ubuntu, Debian, Fedora, Arch)
  • Windows 11 / WSL2

Hardware requirements

  • 1+ CPU cores.
  • 512 MB RAM.
  • 50 MB disk space.

Required software

  • Python 3.11 or newer.
  • pip package manager.
  • virtualenv or venv module.

Free and open-source options

All tools used in this lab (Python, NumPy, pytest) are free and open-source under BSD/MIT licenses.

Installation

python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements/requirements.txt

File structure

  • starter/serving_a_model_over_an_api_lib.py: Student scaffold file.
  • examples/serving_a_model_over_an_api_lib.py: Complete reference implementation.
  • tests/test_serving_a_model_over_an_api_lib.py: Pytest automated validation suite.
  • expected-output/: Verified output logs and baseline values.

How to run

Execute the reference demonstration script:

python3 examples/serving_a_model_over_an_api_lib.py

What the commands do

  • Initializes the serving engine and preloads model weights.
  • Executes single-item prediction with latency profiling.
  • Evaluates output probabilities and fallback states.

Expected output

Serving Demo: Churn Probability = 0.7042, Latency = 0.042ms

Validation steps

  1. Check that predictions before load_model() raise a RuntimeError.
  2. Verify that negative feature values raise a ValueError.
  3. Ensure all unit test assertions pass.

Tests

Run the test runner script:

./tests/run_tests.sh

Cleanup

find . -type d -name "__pycache__" -exec rm -rf {} +
find . -type d -name ".pytest_cache" -exec rm -rf {} +

Troubleshooting

  • Unloaded Model Error: Ensure load_model() is executed prior to calling inference methods.

Security notes

All computations execute locally in memory without external network transmission.

Extension exercises

  1. Implement a Dynamic Batching Queue that flushes after 10ms or 32 items.
  2. Build an automated Prometheus latency counter exporter.

Expected output

FIELDS.md

# Expected Output Fields: Day 194

- `Churn Probability`: Predicted class 1 probability score.
- `Latency ms`: Measured forward pass latency in milliseconds.
- `Used Fallback`: Boolean indicating whether circuit breaker was invoked.

examples-run.txt

Serving Demo: Churn Probability = 0.7042, Latency = 0.042ms

measured-values.txt

Churn Probability: 0.7042
Latency ms: 0.0420
Used Fallback: False

starter-run.txt

Starter scaffold executed. Ready for student implementation.

test-run.txt

============================= test session starts ==============================
collected 2 items

tests/test_serving_a_model_over_an_api_lib.py::test_model_serving_lifecycle_and_predictions PASSED [ 50%]
tests/test_serving_a_model_over_an_api_lib.py::test_serving_validation_and_batch PASSED      [100%]

============================== 2 passed in 0.08s ===============================

Source files

examples/serving_a_model_over_an_api_lib.py (2892 bytes) 
import time
import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Any

@dataclass
class SingleFeatureInput:
    tenure_months: float
    monthly_spend: float
    support_tickets: int

@dataclass
class ServiceHealthStatus:
    is_live: bool
    is_ready: bool
    model_version: str

class ModelServingEngine:
    def __init__(self, model_version: str = "v1.2.0"):
        self.model_version = model_version
        self._is_ready = False
        self._weights = None
        self._bias = 0.0

    def load_model(self, weights: np.ndarray, bias: float) -> None:
        self._weights = np.array(weights, dtype=float)
        self._bias = float(bias)
        self._is_ready = True

    def health_check(self) -> ServiceHealthStatus:
        return ServiceHealthStatus(
            is_live=True,
            is_ready=self._is_ready,
            model_version=self.model_version if self._is_ready else "UNLOADED"
        )

    def _fallback_heuristic(self, features: np.ndarray) -> float:
        spend = features[1]
        tickets = features[2]
        if tickets >= 3 or spend > 150.0:
            return 0.75
        return 0.20

    def predict_single(self, input_data: SingleFeatureInput) -> Dict[str, Any]:
        t0 = time.perf_counter()
        if not self._is_ready:
            raise RuntimeError("Model is not loaded. Service unavailable.")

        if input_data.tenure_months < 0 or input_data.monthly_spend < 0:
            raise ValueError("Feature values cannot be negative")

        features = np.array([
            input_data.tenure_months,
            input_data.monthly_spend,
            float(input_data.support_tickets)
        ])

        try:
            z = float(np.dot(self._weights, features) + self._bias)
            prob = 1.0 / (1.0 + np.exp(-z))
            used_fallback = False
        except Exception:
            prob = self._fallback_heuristic(features)
            used_fallback = True

        latency_ms = (time.perf_counter() - t0) * 1000.0
        return {
            "churn_probability": round(float(prob), 4),
            "prediction": 1 if prob >= 0.5 else 0,
            "used_fallback": used_fallback,
            "model_version": self.model_version,
            "latency_ms": round(latency_ms, 3)
        }

    def predict_batch(self, batch_data: List[SingleFeatureInput]) -> List[Dict[str, Any]]:
        return [self.predict_single(item) for item in batch_data]

def run_serving_demo():
    engine = ModelServingEngine()
    engine.load_model(weights=np.array([0.02, 0.015, 0.40]), bias=-1.5)
    sample = SingleFeatureInput(tenure_months=12.0, monthly_spend=85.0, support_tickets=2)
    res = engine.predict_single(sample)
    print(f"Serving Demo: Churn Probability = {res['churn_probability']}, Latency = {res['latency_ms']}ms")
    return engine, res

if __name__ == "__main__":
    run_serving_demo()
examples/test_serving_a_model_over_an_api_lib.py (1334 bytes) 
import pytest
import numpy as np
from examples.serving_a_model_over_an_api_lib import (
    ModelServingEngine, SingleFeatureInput
)

def test_model_serving_lifecycle_and_predictions():
    engine = ModelServingEngine()
    # Before loading, health is not ready and predict raises error
    h1 = engine.health_check()
    assert h1.is_live is True
    assert h1.is_ready is False

    with pytest.raises(RuntimeError):
        engine.predict_single(SingleFeatureInput(10.0, 50.0, 1))

    # Load weights
    engine.load_model(weights=np.array([0.01, 0.02, 0.5]), bias=-1.0)
    assert engine.health_check().is_ready is True

    # Valid prediction
    res = engine.predict_single(SingleFeatureInput(12.0, 100.0, 3))
    assert "churn_probability" in res
    assert 0.0 <= res["churn_probability"] <= 1.0
    assert res["used_fallback"] is False

def test_serving_validation_and_batch():
    engine = ModelServingEngine()
    engine.load_model(weights=np.array([0.01, 0.02, 0.5]), bias=-1.0)

    # Negative inputs raise ValueError
    with pytest.raises(ValueError):
        engine.predict_single(SingleFeatureInput(-5.0, 50.0, 1))

    # Batch endpoint
    batch = [
        SingleFeatureInput(12.0, 80.0, 0),
        SingleFeatureInput(2.0, 120.0, 4)
    ]
    batch_res = engine.predict_batch(batch)
    assert len(batch_res) == 2
metadata.yml (440 bytes) 
lesson_id: D194
day: 194
kind: lab
languages:
  - python
setup_commands:
  - 'pip install -r requirements/requirements.txt'
run_commands:
  - 'python3 examples/serving_a_model_over_an_api_lib.py'
test_commands:
  - './tests/run_tests.sh'
cleanup_commands:
  - 'find . -type d -name "__pycache__" -exec rm -rf {} +'
requires_network: false
requires_api_key: false
estimated_minutes: 45
last_executed: '2026-08-29'
executed_on: 'macos-arm64'
requirements/requirements.txt (28 bytes) 
numpy>=1.26.0
pytest>=8.0.0
starter/serving_a_model_over_an_api_lib.py (1097 bytes) 
import time
import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Any

@dataclass
class SingleFeatureInput:
    tenure_months: float
    monthly_spend: float
    support_tickets: int

@dataclass
class ServiceHealthStatus:
    is_live: bool
    is_ready: bool
    model_version: str

class ModelServingEngine:
    def __init__(self, model_version: str = "v1.2.0"):
        self.model_version = model_version
        self._is_ready = False
        self._weights = None
        self._bias = 0.0

    def load_model(self, weights: np.ndarray, bias: float) -> None:
        # TODO: Initialize weights and mark ready
        pass

    def health_check(self) -> ServiceHealthStatus:
        # TODO: Return liveness and readiness status
        pass

    def predict_single(self, input_data: SingleFeatureInput) -> Dict[str, Any]:
        # TODO: Implement prediction with validation and fallback circuit breaker
        pass

    def predict_batch(self, batch_data: List[SingleFeatureInput]) -> List[Dict[str, Any]]:
        # TODO: Vectorized batch prediction
        pass
starter/test_serving_a_model_over_an_api_lib.py (1334 bytes) 
import pytest
import numpy as np
from examples.serving_a_model_over_an_api_lib import (
    ModelServingEngine, SingleFeatureInput
)

def test_model_serving_lifecycle_and_predictions():
    engine = ModelServingEngine()
    # Before loading, health is not ready and predict raises error
    h1 = engine.health_check()
    assert h1.is_live is True
    assert h1.is_ready is False

    with pytest.raises(RuntimeError):
        engine.predict_single(SingleFeatureInput(10.0, 50.0, 1))

    # Load weights
    engine.load_model(weights=np.array([0.01, 0.02, 0.5]), bias=-1.0)
    assert engine.health_check().is_ready is True

    # Valid prediction
    res = engine.predict_single(SingleFeatureInput(12.0, 100.0, 3))
    assert "churn_probability" in res
    assert 0.0 <= res["churn_probability"] <= 1.0
    assert res["used_fallback"] is False

def test_serving_validation_and_batch():
    engine = ModelServingEngine()
    engine.load_model(weights=np.array([0.01, 0.02, 0.5]), bias=-1.0)

    # Negative inputs raise ValueError
    with pytest.raises(ValueError):
        engine.predict_single(SingleFeatureInput(-5.0, 50.0, 1))

    # Batch endpoint
    batch = [
        SingleFeatureInput(12.0, 80.0, 0),
        SingleFeatureInput(2.0, 120.0, 4)
    ]
    batch_res = engine.predict_batch(batch)
    assert len(batch_res) == 2
tests/run_tests.sh (227 bytes) 
#!/usr/bin/env bash
set -euo pipefail
echo "========================================"
echo "Running Day 194 Lab Test Suite"
echo "========================================"
pytest tests/ -v
echo "All tests passed successfully."
tests/test_serving_a_model_over_an_api_lib.py (1334 bytes) 
import pytest
import numpy as np
from examples.serving_a_model_over_an_api_lib import (
    ModelServingEngine, SingleFeatureInput
)

def test_model_serving_lifecycle_and_predictions():
    engine = ModelServingEngine()
    # Before loading, health is not ready and predict raises error
    h1 = engine.health_check()
    assert h1.is_live is True
    assert h1.is_ready is False

    with pytest.raises(RuntimeError):
        engine.predict_single(SingleFeatureInput(10.0, 50.0, 1))

    # Load weights
    engine.load_model(weights=np.array([0.01, 0.02, 0.5]), bias=-1.0)
    assert engine.health_check().is_ready is True

    # Valid prediction
    res = engine.predict_single(SingleFeatureInput(12.0, 100.0, 3))
    assert "churn_probability" in res
    assert 0.0 <= res["churn_probability"] <= 1.0
    assert res["used_fallback"] is False

def test_serving_validation_and_batch():
    engine = ModelServingEngine()
    engine.load_model(weights=np.array([0.01, 0.02, 0.5]), bias=-1.0)

    # Negative inputs raise ValueError
    with pytest.raises(ValueError):
        engine.predict_single(SingleFeatureInput(-5.0, 50.0, 1))

    # Batch endpoint
    batch = [
        SingleFeatureInput(12.0, 80.0, 0),
        SingleFeatureInput(2.0, 120.0, 4)
    ]
    batch_res = engine.predict_batch(batch)
    assert len(batch_res) == 2

Troubleshooting

Troubleshooting: Day 194 - Serving a Model over an API

Common Issues

  1. Dimension Mismatch in Forward Pass:
    • Cause: Input feature length does not match weight vector length.
    • Fix: Align feature ordering with weight array dimensionality.

Security notes

Security & Privacy: Day 194 - Serving a Model over an API

Security Guidance

  • All API serving computations execute locally on memory tensors.