Reference library

ML engineering pipelines

Feature prep, batch inference, model-serving hooks, and production ML workflow glue.

12 matches
ML engineering pipelines easy

Build a Mock Random Forest Classifier in Python

Create a simple random-forest-like classifier with random majority voting between trees, including fit, predict, and predict_proba methods.

random forest mock machine learning
Python
import random


class MockRandomForest:
    def __init__(self, n_trees=10, random_state=42):
        self.n_trees = n_trees
        self.random_state = random_state
        self.classes_ = None
        self._class_counts = None
        random.seed(random_state)

    def fit(self, X, y):
        self.classes_ = sorted(…
13 0 Open
ML engineering pipelines easy

Create a Minimal Great Expectations Suite Mock in Python

Build a small Python class that mimics a Great Expectations suite, storing and serializing column expectations as JSON.

great-expectations mock testing
Python
import json


class GreatExpectationsSuite:
    """A minimal mock of a Great Expectations suite."""

    def __init__(self, suite_name, expectations=None):
        self.suite_name = suite_name
        self.expectations = expectations or []

    def add_expectation(self, expectation_type, column=None, kwargs=None):
   …
11 0 Open
ML engineering pipelines medium

Detect Concept Drift in Python with a Simple Statistical Test

Detect concept drift by comparing the mean of recent data against a reference distribution using a z-score-like threshold.

concept drift statistics ml monitoring
Python
import random
import statistics

def detect_drift(recent, reference, threshold=1.5):
    ref_mean = statistics.mean(reference)
    ref_std = statistics.stdev(reference)
    
    recent_mean = statistics.mean(recent)
    drift_score = abs(recent_mean - ref_mean) / (ref_std if ref_std > 0 else 1)
    
    drifted = drif…
15 0 Open
ML engineering pipelines easy

How to Load CSV Training Data in Python Without Pandas

Load CSV training data using Python's standard library and mock it with io.StringIO for testing, returning headers and rows as dictionaries.

csv ml-pipelines io-stringio
Python
import csv
from pathlib import Path


def load_csv_training_data(file_path: str | Path) -> tuple[list[str], list[dict[str, str]]]:
    """Load CSV training data and return headers plus rows as dictionaries."""
    with open(file_path, mode="r", newline="", encoding="utf-8") as csv_file:
        reader = csv.DictReader…
14 0 Open
ML engineering pipelines medium

How to Mock MLflow Model Registration in Python

Build a lightweight in-memory mock of MLflow's MlflowClient to test model registration, versioning, and stage transitions without a tracking server.

mlflow mocking model-registry
Python
from mlflow.tracking import MlflowClient
from mlflow.entities import ModelVersion, Model


class MockMlflowClient:
    """Minimal mock of MlflowClient's model registration methods."""
    
    def __init__(self):
        self.registered_models = {}
        self.model_versions = {}
    
    def register_model(self, mod…
14 0 Open
ML engineering pipelines easy

How to Mock MLflow log_params and log_metrics in Python

Use unittest.mock to patch MLflow's log_param and log_metric, run the training function, and verify logging calls without touching a real tracking server.

mlflow mock testing
Python
from unittest.mock import Mock, patch
import mlflow


def train_model():
    mlflow.log_param("learning_rate", 0.01)
    mlflow.log_param("epochs", 10)
    mlflow.log_metric("accuracy", 0.95)
    mlflow.log_metric("loss", 0.05)
    return "Training completed"


if __name__ == "__main__":
    with patch("mlflow.log_par…
15 0 Open
ML engineering pipelines medium

How to Mock ROC AUC in Python

Compute ROC AUC from scratch in Python using pairwise comparisons between positive and negative score distributions, ideal for testing ML models without sklearn.

machine-learning model-evaluation auc
Python
import random
from math import comb


def mock_roc_auc(scores, labels):
    """Compute mock ROC AUC by simulating a classifier's score distribution."""
    random.seed(42)
    n = len(labels)
    pos_scores = [scores[i] for i in range(n) if labels[i] == 1]
    neg_scores = [scores[i] for i in range(n) if labels[i] == …
12 0 Open
ML engineering pipelines easy

How to Mock train_test_split in Python for Unit Testing

Build a lightweight mock of sklearn's train_test_split to unit test ML pipeline code without needing the full library or deterministic random state.

train_test_split mock unit-testing
Python
import numpy as np
from sklearn.model_selection import train_test_split
from unittest.mock import patch

def mock_train_test_split(X, y, test_size=0.25, random_state=None, **kwargs):
    """A simple mock implementation of train_test_split."""
    n_samples = len(X)
    n_test = int(n_samples * test_size)
    n_train =…
12 0 Open
ML engineering pipelines medium

How to Train a Gradient Boosting Regressor in Python

Build and evaluate a scikit-learn GradientBoostingRegressor on a synthetic dataset, printing test MSE and feature importances.

sklearn gradient-boosting regression
Python
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.metrics import mean_squared_error

def train_gradient_boosting_mock():
    # Toy regression dataset
    np.random.seed(42)
    X = np.random.rand(100, 3) * 10
    y = 2 * X[:, 0] - 1.5 * X[:, 1] + 0.5 * X[:, 2] + np.random.normal(0,…
13 0 Open
ML engineering pipelines easy

How to implement a canary traffic split in Python

Route incoming traffic between stable and canary model or service versions using a weight-based random split with deterministic testing.

canary traffic-split random
Python
import random


def canary_route(service_name: str, canary_weight: float = 0.2) -> str:
    """Route traffic between stable and canary versions based on weight."""
    rng = random.Random(42)  # deterministic for reproducible demo
    if rng.random() < canary_weight:
        return f"{service_name}-canary"
    return …
14 0 Open
ML engineering pipelines medium

How to mock an artifact store with local paths in Python for ML pipelines

Create a temporary local artifact store with dummy files and metadata to test ML pipeline code without real storage.

ml-pipelines mock tempfile
Python
import tempfile
from pathlib import Path
import json


def create_artifact_store_mock(base_path: Path = None):
    """Create a local artifact store mock directory structure."""
    if base_path is None:
        base_path = Path(tempfile.mkdtemp())

    store_layout = {
        "artifacts": [
            {"name": "mode…
13 0 Open
ML engineering pipelines easy

Model registry version mock in Python

A simple in-memory model registry that stores model versions with metadata and supports version listing and latest retrieval.

ml-engineering model-registry versioning
Python
class ModelRegistry:
    def __init__(self):
        self.models = {}

    def register(self, name, version, model_type, metrics=None):
        if name not in self.models:
            self.models[name] = []
        entry = {
            "version": version,
            "model_type": model_type,
            "metrics": m…
13 0 Open

Browse by section

Each section groups closely related Python snippets.

ML engineering pipelines — Python code examples

What you will find here

This page collects ml engineering pipelines snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.

Samples vs tutorials and challenges

Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.