ML engineering pipelines
Feature prep, batch inference, model-serving hooks, and production ML workflow glue.
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.
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…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.
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…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.
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] == …How to Mock Shadow Mode Inference in Python
Simulates running multiple candidate models in shadow mode by adding randomized delays and returning their outputs alongside a primary model's output.
import random
import time
def shadow_mode_inference(candidates, mock_delay=0.1):
"""
Simulates running multiple candidate models in 'shadow mode'
by adding tiny randomized delays and returning their outputs
alongside the primary model's output.
"""
primary_output = "primary: answer"
shado…How to Mock a Kubeflow Pipeline in Python
Build a minimal in-memory mock of a Kubeflow pipeline DAG using dataclasses and OrderedDict to chain component functions.
from typing import Dict, Any
from dataclasses import dataclass, field
from collections import OrderedDict
@dataclass
class KubeflowPipelineMock:
"""A minimal mock of a Kubeflow pipeline DAG."""
name: str
components: OrderedDict[str, callable] = field(default_factory=OrderedDict)
def add_component(se…How to Run Batch Predictions with a Mock Model in Python
Build a lightweight mock model class and run predictions across a batch of samples, returning results as a plain Python list.
import numpy as np
class MockModel:
def __init__(self, weights):
self.weights = np.array(weights)
def predict(self, X):
return X @ self.weights
def predict_batch(model, batch):
"""Run predictions for a batch of samples and return results as a list."""
return model.predict(np.array(ba…How to Save and Load a Mock Model with Pickle and joblib in Python
Serialize a custom machine learning model to a .joblib file with joblib.dump, reload it, and run a prediction with joblib.load.
import joblib
from pathlib import Path
class MockModel:
def __init__(self, weights):
self.weights = weights
def predict(self, features):
return sum(w * f for w, f in zip(self.weights, features))
def save_model_pickle(model, filepath):
with open(filepath, "wb") as f:
joblib.dump(…How to Simulate an Airflow ML Pipeline in Python
Mock an Airflow ML pipeline in plain Python by defining steps, simulating their execution with delays, and returning a success summary.
from datetime import datetime, timedelta
import time
class MLPipeline:
def __init__(self, pipeline_name):
self.pipeline_name = pipeline_name
self.steps = []
def add_step(self, step_name, duration_seconds):
self.steps.append({"name": step_name, "duration": duration_seconds})
def …How to Stage ML Model Workflows with Python Classes
Defines a Stage class to model ML pipeline stages with variants and mocks, printing grammar for Model, Staging, and Production stages.
class Stage:
def __init__(self, name):
self.name = name
self.mocks = []
self.variants = []
def add_mock(self, mock_name):
self.mocks.append(mock_name)
def add_variant(self, variant_name, productions=()):
self.variants.append((variant_name, list(productions)))
…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.
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,…How to Trigger Model Retraining on Drift in Python
Automatically detects accuracy drift in a mock ML model and triggers retraining when performance falls below a threshold.
import random
import time
class MockModel:
def __init__(self, name):
self.name = name
self.accuracy = 0.85
self.version = 1
def train(self, data_size):
# Simulate training time and accuracy improvement
time.sleep(0.1)
drift = random.uniform(-0.02, 0.02)
…How to do feature selection with VarianceThreshold in Python
This code demonstrates how to use scikit-learn's VarianceThreshold to remove low-variance features from a NumPy array, keeping only those that vary enough to be useful for modeling.
import numpy as np
from sklearn.feature_selection import VarianceThreshold
def main():
# Mock dataset: 4 samples, 5 features
X = np.array([
[0.1, 0.2, 1.0, 1.0, 0.5],
[0.2, 0.2, 0.0, 1.0, 0.4],
[0.1, 0.2, 1.0, 1.0, 0.6],
[0.3, 0.2, 1.0, 0.0, 0.5]
])
# Select features w…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.
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 …K-Fold Cross Validation in Python: A Simple Implementation
Implements k-fold cross validation from scratch, splitting data into folds and computing MSE scores for a baseline mean-predictor model.
import random
from statistics import mean
def cross_validation_scores(data, labels, k=5, seed=42):
random.seed(seed)
indices = list(range(len(data)))
random.shuffle(indices)
fold_size = len(indices) // k
folds = []
for i in range(k):
if i == k - 1:
folds.append(indices[i *…Load CSV Training Data Without Pandas in Python
This code loads a CSV file into a list of dictionaries using only the standard library, ideal for small ML training data without heavy dependencies.
import csv
from pathlib import Path
def load_csv(path):
"""Load CSV file into list of dicts without pandas."""
rows = []
with open(path, newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
rows.append(dict(row))
return rows
if __name__ == "__m…Mock a Flyte ML workflow in Python
Build a lightweight mock of a Flyte ML pipeline with dataclasses and a simple execution loop that passes outputs between tasks.
from dataclasses import dataclass, field
from typing import List, Dict, Optional
import time
@dataclass
class FlyteTask:
name: str
inputs: Dict = field(default_factory=dict)
outputs: Dict = field(default_factory=dict)
def run(self) -> Dict:
time.sleep(0.1) # simulate work
return sel…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.
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…Train Logistic Regression From Scratch in Python
Trains a binary logistic regression model using gradient descent on mock data, printing learned weights and probabilities.
import numpy as np
# Mock data: 2 features, binary classification
X = np.array([[1, 2], [2, 3], [3, 4], [4, 5], [5, 6]])
y = np.array([0, 0, 1, 1, 1])
# Add bias term (column of ones)
X_b = np.c_[np.ones((X.shape[0], 1)), X]
# Initialize parameters
theta = np.zeros(X_b.shape[1])
# Hyperparameters
learning_rate = 0…Training Pipeline Orchestration Mock DAG in Python
Build a mock DAG orchestrator that runs ML pipeline stages in dependency order using topological sorting (Kahn's algorithm).
from collections import deque
from dataclasses import dataclass, field
@dataclass
class DAGNode:
name: str
task: callable
dependencies: list[str] = field(default_factory=list)
class MockDAG:
def __init__(self, nodes: list[DAGNode]):
self.nodes = {n.name: n for n in nodes}
self.execu…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.