How to Deploy Your First ML Model with Flask or FastAPI (Beginner)

import joblib
import json
from pathlib import Path
from datetime import datetime


def save_sklearn_model(
    model, scaler, feature_names, output_dir='model_artifacts',
    model_version='1.0.0'
):
    """Save sklearn model with its complete preprocessing pipeline.

    The model, scaler, and metadata are saved as a single versioned
    artifact directory. All three files must travel together — a model
    without its scaler produces garbage predictions on raw input.

    Args:
        model: fitted sklearn estimator
        scaler: fitted sklearn transformer (StandardScaler, etc.)
        feature_names: list of feature name strings in training order
        output_dir: directory to save artifacts
        model_version: semantic version string for tracking
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    # Save the trained model
    joblib.dump(model, output_path / 'model.joblib')

    # Save the fitted scaler — this MUST travel with the model
    # Without it, the API receives raw features and the model
    # expects scaled features. Predictions are silently wrong.
    joblib.dump(scaler, output_path / 'scaler.joblib')

    # Save metadata for debugging and validation at load time
    import sklearn
    metadata = {
        'model_type': type(model).__name__,
        'model_version': model_version,
        'feature_names': feature_names,
        'n_features': len(feature_names),
        'sklearn_version': sklearn.__version__,
        'python_version': __import__('sys').version,
        'saved_at': datetime.utcnow().isoformat(),
        'training_notes': 'Add any relevant notes about training data or parameters here'
    }
    with open(output_path / 'metadata.json', 'w') as f:
        json.dump(metadata, f, indent=2)

    print(f"Model artifacts saved to {output_path}/")
    for file in sorted(output_path.iterdir()):
        size_kb = file.stat().st_size / 1024
        print(f"  {file.name}: {size_kb:.1f} KB")


def save_tensorflow_model(model, output_dir='model_artifacts/tf_model'):
    """Save TensorFlow model in SavedModel format.

    SavedModel is more portable across TF versions than HDF5 (.h5)
    and includes the computation graph, making it suitable for
    TensorFlow Serving and TFLite conversion.
    """
    model.save(output_dir)
    print(f"TensorFlow SavedModel saved to {output_dir}/")


# --- Example usage ---
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
import numpy as np

# Simulate training workflow
np.random.seed(42)
X = np.random.randn(1000, 3)
y = (X[:, 0] + X[:, 1] > 0).astype(int)

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# Fit preprocessing and model
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)

model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train_scaled, y_train)

# Evaluate before saving
X_test_scaled = scaler.transform(X_test)
accuracy = model.score(X_test_scaled, y_test)
print(f"Test accuracy: {accuracy:.3f}")

# Save everything together
save_sklearn_model(
    model=model,
    scaler=scaler,
    feature_names=['age', 'income', 'credit_score'],
    model_version='1.0.0'
)

from flask import Flask, request, jsonify
import joblib
import numpy as np
import json
import logging
from pathlib import Path

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)

app = Flask(__name__)

# ---------------------------------------------------------------
# LOAD MODEL AT MODULE LEVEL — not inside a route function.
# This code runs once when the process starts.
# The loaded objects persist in memory across all requests.
# ---------------------------------------------------------------
MODEL_DIR = Path(__file__).parent / 'model_artifacts'

try:
    model = joblib.load(MODEL_DIR / 'model.joblib')
    scaler = joblib.load(MODEL_DIR / 'scaler.joblib')
    with open(MODEL_DIR / 'metadata.json') as f:
        metadata = json.load(f)
    logger.info(
        f"Model loaded: {metadata['model_type']}, "
        f"{metadata['n_features']} features, "
        f"sklearn {metadata.get('sklearn_version', 'unknown')}"
    )
except Exception as e:
    logger.error(f"Failed to load model: {e}")
    raise  # Fail fast — do not start serving with no model


@app.route('/health', methods=['GET'])
def health():
    """Health check for load balancers and container orchestrators.

    Returns 200 only when the model is loaded and ready.
    Kubernetes liveness probes and Docker HEALTHCHECK use this.
    """
    return jsonify({
        'status': 'healthy',
        'model_type': metadata['model_type'],
        'model_version': metadata.get('model_version', 'unknown'),
        'n_features': metadata['n_features']
    })


@app.route('/predict', methods=['POST'])
def predict():
    """Prediction endpoint.

    Expects JSON body: {"features": [value1, value2, ...]}
    Returns: {"prediction": [...], "probabilities": [[...]], "model_version": "..."}
    """
    try:
        data = request.get_json(force=True)

        # --- Input validation ---
        if data is None:
            return jsonify({'error': 'Request body must be valid JSON'}), 400

        if 'features' not in data:
            return jsonify({'error': 'Missing "features" key in request body'}), 400

        features_raw = data['features']

        if not isinstance(features_raw, list):
            return jsonify({'error': '"features" must be a list of numbers'}), 400

        # Check for null/None values
        if any(v is None for v in features_raw):
            return jsonify({'error': '"features" contains null values'}), 400

        features = np.array(features_raw, dtype=np.float64)

        # Check for NaN or Inf
        if not np.isfinite(features).all():
            return jsonify({'error': '"features" contains NaN or Infinity values'}), 400

        # Validate feature count
        expected = metadata['n_features']
        if features.ndim == 1:
            features = features.reshape(1, -1)
        if features.shape[1] != expected:
            return jsonify({
                'error': f'Expected {expected} features, got {features.shape[1]}',
                'expected_features': metadata.get('feature_names', [])
            }), 400

        # --- Preprocess and predict ---
        features_scaled = scaler.transform(features)
        prediction = model.predict(features_scaled)

        response = {
            'prediction': prediction.tolist(),
            'model_version': metadata.get('model_version', 'unknown')
        }

        # Include probabilities if the model supports them
        if hasattr(model, 'predict_proba'):
            probabilities = model.predict_proba(features_scaled)
            response['probabilities'] = probabilities.tolist()

        return jsonify(response)

    except ValueError as e:
        logger.warning(f"Validation error: {e}")
        return jsonify({'error': f'Invalid input: {str(e)}'}), 400
    except Exception as e:
        logger.error(f"Prediction failed: {e}", exc_info=True)
        return jsonify({'error': 'Internal server error'}), 500


if __name__ == '__main__':
    # Development only — never use this in production
    app.run(host='0.0.0.0', port=5000, debug=False)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, field_validator
import joblib
import numpy as np
import json
import logging
from pathlib import Path
from typing import List, Optional

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)

app = FastAPI(
    title='ML Prediction API',
    description='Scikit-learn model serving endpoint with automatic input validation',
    version='1.0.0'
)

# ---------------------------------------------------------------
# LOAD MODEL AT STARTUP — same pattern as Flask
# ---------------------------------------------------------------
MODEL_DIR = Path(__file__).parent / 'model_artifacts'

try:
    model = joblib.load(MODEL_DIR / 'model.joblib')
    scaler = joblib.load(MODEL_DIR / 'scaler.joblib')
    with open(MODEL_DIR / 'metadata.json') as f:
        metadata = json.load(f)
    logger.info(
        f"Model loaded: {metadata['model_type']}, "
        f"{metadata['n_features']} features"
    )
except Exception as e:
    logger.error(f"Failed to load model: {e}")
    raise


# ---------------------------------------------------------------
# PYDANTIC MODELS — define input/output schemas once
# FastAPI validates every request against these automatically.
# Invalid requests get a clear 422 error before your code runs.
# ---------------------------------------------------------------
class PredictionRequest(BaseModel):
    """Input schema for prediction requests."""
    features: List[float] = Field(
        ...,
        description='List of numeric feature values in training order',
        min_length=1
    )

    @field_validator('features')
    @classmethod
    def validate_features(cls, v):
        """Reject NaN, Infinity, and wrong feature count."""
        import math
        for i, val in enumerate(v):
            if math.isnan(val) or math.isinf(val):
                raise ValueError(
                    f'Feature at index {i} is {val} — must be a finite number'
                )

        expected = metadata['n_features']
        if len(v) != expected:
            raise ValueError(
                f'Expected {expected} features, got {len(v)}. '
                f'Expected order: {metadata.get("feature_names", [])}'
            )
        return v


class PredictionResponse(BaseModel):
    """Output schema for prediction responses."""
    prediction: List[int]
    probabilities: Optional[List[List[float]]] = None
    model_version: str


class HealthResponse(BaseModel):
    """Output schema for health check."""
    status: str
    model_type: str
    model_version: str
    n_features: int


# ---------------------------------------------------------------
# ENDPOINTS
# ---------------------------------------------------------------
@app.get('/health', response_model=HealthResponse)
def health():
    """Health check for load balancers and orchestrators."""
    return HealthResponse(
        status='healthy',
        model_type=metadata['model_type'],
        model_version=metadata.get('model_version', 'unknown'),
        n_features=metadata['n_features']
    )


@app.post('/predict', response_model=PredictionResponse)
def predict(request: PredictionRequest):
    """Run model inference on the provided features.

    Input validation (type checking, NaN detection, feature count)
    is handled automatically by the Pydantic model above.
    By the time this function runs, the input is guaranteed valid.
    """
    try:
        features = np.array(request.features).reshape(1, -1)
        features_scaled = scaler.transform(features)
        prediction = model.predict(features_scaled)

        response = PredictionResponse(
            prediction=prediction.tolist(),
            model_version=metadata.get('model_version', 'unknown')
        )

        if hasattr(model, 'predict_proba'):
            probabilities = model.predict_proba(features_scaled)
            response.probabilities = probabilities.tolist()

        return response

    except Exception as e:
        logger.error(f"Prediction failed: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail='Internal server error')


# ---------------------------------------------------------------
# Run with: uvicorn fastapi_app:app --host 0.0.0.0 --port 8000 --workers 4
# Auto-generated docs available at: http://localhost:8000/docs
# ---------------------------------------------------------------

import numpy as np
from typing import List, Dict, Any, Optional


class InputValidator:
    """Validate prediction inputs before they reach the model.

    Every check that fails here is a prediction error prevented.
    A model will happily predict on a credit score of 99999 or
    an age of -5. The validator catches these before the model sees them.
    """

    def __init__(self, metadata: dict, feature_ranges: Optional[dict] = None):
        self.n_features = metadata['n_features']
        self.feature_names = metadata.get('feature_names', [])
        self.feature_ranges = feature_ranges or {}

    def validate(self, features: List[float]) -> np.ndarray:
        """Validate and convert input features.

        Returns a numpy array ready for preprocessing.
        Raises ValueError with a clear message if validation fails.
        """
        # Check type
        if not isinstance(features, list):
            raise ValueError(
                f'"features" must be a list, got {type(features).__name__}'
            )

        # Check length
        if len(features) != self.n_features:
            raise ValueError(
                f'Expected {self.n_features} features, got {len(features)}. '
                f'Expected: {self.feature_names}'
            )

        # Check for null/None values
        for i, val in enumerate(features):
            if val is None:
                name = self.feature_names[i] if i < len(self.feature_names) else f'index {i}'
                raise ValueError(f'Feature "{name}" is null')

        # Convert to numpy and check for NaN/Inf
        arr = np.array(features, dtype=np.float64)
        if not np.isfinite(arr).all():
            bad_indices = np.where(~np.isfinite(arr))[0]
            bad_names = [
                self.feature_names[i] if i < len(self.feature_names) else f'index {i}'
                for i in bad_indices
            ]
            raise ValueError(
                f'Non-finite values in features: {bad_names}. '
                f'Values must be finite numbers (no NaN, no Infinity).'
            )

        # Check feature ranges (domain validation)
        for feature_name, (min_val, max_val) in self.feature_ranges.items():
            if feature_name in self.feature_names:
                idx = self.feature_names.index(feature_name)
                val = arr[idx]
                if val < min_val or val > max_val:
                    raise ValueError(
                        f'Feature "{feature_name}" value {val} is outside '
                        f'expected range [{min_val}, {max_val}]. '
                        f'This may indicate a data pipeline error.'
                    )

        return arr.reshape(1, -1)


# --- Example usage ---
validator = InputValidator(
    metadata={'n_features': 3, 'feature_names': ['age', 'income', 'credit_score']},
    feature_ranges={
        'age': (0, 120),
        'income': (0, 10_000_000),
        'credit_score': (300, 850)
    }
)

# These will raise clear ValueError messages:
# validator.validate([25, 50000])              # Wrong count
# validator.validate([25, 50000, None])         # Null value
# validator.validate([25, 50000, float('nan')]) # NaN value
# validator.validate([-5, 50000, 720])          # Age out of range
# validator.validate([25, 50000, 8500])         # Credit score out of range (typo)

# This will pass:
result = validator.validate([35, 75000, 720])
print(f'Validated input shape: {result.shape}')  # (1, 3)

# Dockerfile for ML API deployment
# Uses a slim Python base image to minimize attack surface and image size
FROM python:3.11-slim

# Set working directory inside the container
WORKDIR /app

# Install system dependencies (if needed for scipy, numpy, etc.)
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies first — separate layer for Docker cache
# When you change only application code, this layer is cached
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY app/ ./app/

# Copy model artifacts
# For models >500MB, consider downloading from S3/GCS at startup instead
COPY model_artifacts/ ./model_artifacts/

# Create a non-root user for security
RUN useradd --create-home appuser
USER appuser

# Health check — container orchestrators use this to determine readiness
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# Expose the port (documentation — does not actually publish the port)
EXPOSE 8000

# Run with production ASGI server
# --workers 2: spawn 2 processes (adjust based on container CPU allocation)
# --timeout 120: allow up to 120s for model loading at startup
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2", "--timeout-keep-alive", "120"]

import pytest
import numpy as np
import json
from pathlib import Path

# FastAPI test client (for Flask, use app.test_client())
from fastapi.testclient import TestClient
from app.main import app, model, scaler, metadata

client = TestClient(app)


class TestHealthEndpoint:
    """Verify the health check works before testing predictions."""

    def test_health_returns_200(self):
        response = client.get('/health')
        assert response.status_code == 200
        data = response.json()
        assert data['status'] == 'healthy'
        assert 'model_type' in data
        assert data['n_features'] == metadata['n_features']


class TestInputValidation:
    """Verify that invalid inputs are rejected with clear errors."""

    def test_empty_body_returns_422(self):
        response = client.post('/predict', json={})
        assert response.status_code == 422  # Pydantic validation error

    def test_missing_features_key_returns_422(self):
        response = client.post('/predict', json={'data': [1, 2, 3]})
        assert response.status_code == 422

    def test_wrong_feature_count_returns_422(self):
        response = client.post('/predict', json={'features': [1.0, 2.0]})
        assert response.status_code == 422

    def test_nan_features_returns_422(self):
        n = metadata['n_features']
        features = [float('nan')] + [1.0] * (n - 1)
        response = client.post('/predict', json={'features': features})
        assert response.status_code == 422

    def test_null_features_returns_422(self):
        n = metadata['n_features']
        features = [None] + [1.0] * (n - 1)
        response = client.post('/predict', json={'features': features})
        assert response.status_code == 422

    def test_string_features_returns_422(self):
        response = client.post('/predict', json={'features': ['a', 'b', 'c']})
        assert response.status_code == 422


class TestPredictEndpoint:
    """Verify that valid inputs produce correct responses."""

    def test_valid_input_returns_200_with_prediction(self):
        n = metadata['n_features']
        features = [35.0, 75000.0, 720.0][:n]
        response = client.post('/predict', json={'features': features})
        assert response.status_code == 200
        data = response.json()
        assert 'prediction' in data
        assert 'model_version' in data
        assert isinstance(data['prediction'], list)

    def test_probabilities_sum_to_one(self):
        n = metadata['n_features']
        features = [35.0, 75000.0, 720.0][:n]
        response = client.post('/predict', json={'features': features})
        data = response.json()
        if 'probabilities' in data and data['probabilities']:
            prob_sum = sum(data['probabilities'][0])
            assert abs(prob_sum - 1.0) < 0.01, f'Probabilities sum to {prob_sum}'


class TestPredictionConsistency:
    """THE MOST IMPORTANT TEST CLASS.

    Verify API predictions match direct model predictions on identical input.
    This catches preprocessing mismatches that produce valid HTTP 200 responses
    with wrong predictions — the most dangerous class of deployment bugs.
    """

    @pytest.fixture
    def sample_inputs(self):
        """Multiple test cases to cover different input ranges."""
        n = metadata['n_features']
        return [
            [35.0, 75000.0, 720.0][:n],
            [22.0, 30000.0, 580.0][:n],
            [65.0, 150000.0, 800.0][:n],
            [0.0, 0.0, 300.0][:n],  # Edge case: minimum values
        ]

    def test_api_matches_direct_model(self, sample_inputs):
        """Run identical input through both API and raw model."""
        for features in sample_inputs:
            # API prediction
            response = client.post('/predict', json={'features': features})
            assert response.status_code == 200
            api_prediction = response.json()['prediction']

            # Direct model prediction with same preprocessing
            raw_input = np.array(features).reshape(1, -1)
            scaled_input = scaler.transform(raw_input)
            model_prediction = model.predict(scaled_input).tolist()

            assert api_prediction == model_prediction, (
                f'Consistency check failed for input {features}: '
                f'API returned {api_prediction}, '
                f'direct model returned {model_prediction}'
            )

import logging
import time
import json
from collections import Counter, deque
from datetime import datetime, timezone
from typing import Any, Optional, List

logger = logging.getLogger('ml_api')


class PredictionMonitor:
    """Track prediction metrics for production monitoring.

    Collects latency, error rates, and prediction distribution.
    Expose via a /metrics endpoint for Prometheus scraping or
    a /stats endpoint for manual inspection.

    In production, replace this with proper observability tools
    (Prometheus + Grafana, Datadog, or OpenTelemetry). This class
    demonstrates the metrics you should be collecting regardless
    of the tool you use.
    """

    def __init__(self, max_latency_samples: int = 10000):
        self.total_requests = 0
        self.error_count = 0
        self.latency_ms: deque = deque(maxlen=max_latency_samples)
        self.prediction_distribution = Counter()
        self._start_time = datetime.now(timezone.utc)

    def log_request(
        self,
        features_summary: dict,
        prediction: Any,
        latency_ms: float,
        status: str = 'success'
    ):
        """Log a single prediction request.

        Args:
            features_summary: non-PII summary of input
                              (e.g., feature count, value ranges)
            prediction: the model's output
            latency_ms: wall-clock inference time in milliseconds
            status: 'success' or 'error'
        """
        self.total_requests += 1
        self.latency_ms.append(latency_ms)

        if status == 'error':
            self.error_count += 1

        if prediction is not None:
            pred_key = str(prediction)
            self.prediction_distribution[pred_key] += 1

        # Structured JSON log — ingestible by log aggregation systems
        logger.info(json.dumps({
            'event': 'prediction',
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'status': status,
            'latency_ms': round(latency_ms, 2),
            'prediction': prediction,
            'feature_count': features_summary.get('count', 0),
        }))

    def get_stats(self) -> dict:
        """Return summary statistics for the /stats endpoint."""
        if not self.latency_ms:
            return {
                'message': 'No requests recorded yet',
                'uptime_seconds': (
                    datetime.now(timezone.utc) - self._start_time
                ).total_seconds()
            }

        sorted_latencies = sorted(self.latency_ms)
        n = len(sorted_latencies)

        return {
            'total_requests': self.total_requests,
            'error_count': self.error_count,
            'error_rate_pct': round(
                self.error_count / max(self.total_requests, 1) * 100, 2
            ),
            'latency_p50_ms': round(sorted_latencies[n // 2], 2),
            'latency_p95_ms': round(sorted_latencies[int(n * 0.95)], 2),
            'latency_p99_ms': round(sorted_latencies[int(n * 0.99)], 2),
            'prediction_distribution': dict(
                self.prediction_distribution.most_common(20)
            ),
            'uptime_seconds': round(
                (datetime.now(timezone.utc) - self._start_time).total_seconds()
            )
        }


# Singleton instance — import and use across the application
monitor = PredictionMonitor()


# --- Example integration with FastAPI ---
# @app.post('/predict')
# def predict(request: PredictionRequest):
#     start = time.perf_counter()
#     try:
#         prediction = model.predict(preprocessed)
#         latency = (time.perf_counter() - start) * 1000
#         monitor.log_request(
#             features_summary={'count': len(request.features)},
#             prediction=prediction.tolist(),
#             latency_ms=latency
#         )
#         return ...
#     except Exception as e:
#         latency = (time.perf_counter() - start) * 1000
#         monitor.log_request(
#             features_summary={'count': len(request.features)},
#             prediction=None,
#             latency_ms=latency,
#             status='error'
#         )
#         raise
#
# @app.get('/stats')
# def stats():
#     return monitor.get_stats()