Reference library

Python Code Samples

Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.

166 matches
Microservices patterns easy

How to Order Partition Key Events in Python (Mock Stream)

Generate a mock event stream grouped by partition key and sort it deterministically by key then sequence in Python.

partition events sorting
Python
import itertools
import random


def partition_key_events(keys, events_per_key=3, seed=None):
    """Produce a realistic-looking, but mock, event stream grouped by partition key.

    Args:
        keys: iterable of partition keys (e.g. strings or ints).
        events_per_key: how many events we want per key.
       …
11 0 Open
Big data & Spark easy

How to Broadcast a Small Lookup Table in Python

Simulates broadcasting a small lookup table by iterating key-value pairs and emitting packed rows to subscribers with deterministic output.

broadcast lookup-table dictionary
Python
import random

# Generate a deterministic mock broadcast of a small lookup table
# with 5 keys and random integer values (seeded for reproducibility)

data = {
    "sensor_a": 22,
    "sensor_b": 87,
    "sensor_c": 43,
    "sensor_d": 65,
    "sensor_e": 31,
}

# Simulate a broadcast to subscribers by iterating and p…
15 0 Open
Big data & Spark medium

How to Mock a UDAF Aggregate Function in Python

This code provides a minimal mock of a User-Defined Aggregate Function (UDAF), simulating the initialize-update-merge-finalize lifecycle with a defaultdict counter.

udaf aggregate mock
Python
from collections import defaultdict

class MockUDAF:
    """A minimal mock of a User-Defined Aggregate Function.

    Simulates aggregate lifecycle: initialize, update per row,
    and finalize the result.
    """

    def __init__(self):
        self._buffer = defaultdict(int)

    def initialize(self):
        """Re…
13 0 Open
ML engineering pipelines medium

Bayesian Optimization in Python: A Simplified Mock Implementation

A toy Bayesian optimization loop with a Gaussian process prior, expected improvement acquisition, and noisy sampling to find a function's minimum.

bayesian-optimization gaussian-process hyperparameter-tuning
Python
import random
import math

class BayesianOptimizer:
    def __init__(self, noise=0.1):
        self.noise = noise
        self.observations = []
    
    def objective(self, x):
        return (math.sin(3*x) + 0.5*x) / (1 + x**2)
    
    def gaussian_process_prior(self, x1, x2, length_scale=0.5):
        return math.…
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):
   …
13 0 Open
ML engineering pipelines medium

How to Build a Mock ML Pipeline with Prefect in Python

Create a lightweight Prefect flow with mock preprocessing, training, and evaluation tasks to prototype an ML pipeline end-to-end.

prefect machine-learning pipeline
Python
from prefect import task, flow
from datetime import datetime


@task
def preprocess_data(raw_value: float) -> float:
    """Mock preprocessing: normalize the input value."""
    return raw_value / 100.0


@task
def train_model(features: float) -> dict:
    """Mock training: return a fake model artifact."""
    return …
12 0 Open
ML engineering pipelines easy

How to Build a Simple ML Pipeline with ZenML in Python

Build a mock machine learning pipeline with ZenML steps for data loading, training, and evaluation, and run it to print the final accuracy.

zenml ml pipeline
Python
from zenml import pipeline, step


@step
def load_data() -> dict:
    """Simulate loading data from a source."""
    return {"accuracy": 0.0, "loss": 1.0}


@step
def train_model(data: dict) -> dict:
    """Simulate training a model."""
    data["accuracy"] = 0.95
    data["loss"] = 0.1
    return data


@step
def eva…
13 0 Open
ML engineering pipelines easy

How to Create a Mock Metaflow Flow in Python

Build a minimal Metaflow flow with two sequential steps that pass data between them using instance attributes.

metaflow ml-pipelines workflow
Python
from metaflow import FlowSpec, step, current


class MockFlow(FlowSpec):
    """A minimal Metaflow flow to demonstrate basic steps and branching."""

    @step
    def start(self):
        self.category = "mock"
        print(f"Start step for {self.category} flow")
        self.next(self.process)

    @step
    def pr…
15 0 Open
ML engineering pipelines medium

How to Create a Mock ONNX Model in Python

Build and export a minimal mock ONNX model with a Reshape and Gemm layer using the onnx helper API.

onnx model-export mlops
Python
import onnx
import numpy as np
from onnx import helper, TensorProto

def create_mock_model():
    # Define input and output tensors
    input_tensor = helper.make_tensor_value_info('input', TensorProto.FLOAT, [1, 3, 224, 224])
    output_tensor = helper.make_tensor_value_info('output', TensorProto.FLOAT, [1, 10])

   …
16 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 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 a Kubeflow Pipeline in Python

Build a minimal in-memory mock of a Kubeflow pipeline DAG using dataclasses and OrderedDict to chain component functions.

kubeflow pipelines mlops
Python
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…
14 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 easy

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.

airflow ml pipeline
Python
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 …
13 0 Open
ML engineering pipelines easy

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.

ml drift-detection retraining
Python
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)
       …
16 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 easy

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.

csv data-loading standard-library
Python
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…
13 0 Open
ML engineering pipelines medium

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).

dag pipeline topological-sort
Python
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…
13 0 Open
A/B testing & experimentation medium

Benjamini Hochberg FDR Correction in Python

Implement the Benjamini-HHochberg false discovery rate (FDR) procedure in Python to control the expected proportion of false positives among rejected hypotheses.

fdr multiple testing hypothesis testing
Python
import numpy as np

def benjamini_hochberg(p_values, alpha=0.05):
    p_values = np.array(p_values)
    n = len(p_values)
    sorted_idx = np.argsort(p_values)
    sorted_p = p_values[sorted_idx]
    
    thresholds = (np.arange(1, n + 1) / n) * alpha
    significant = sorted_p <= thresholds
    
    if not significan…
16 0 Open
A/B testing & experimentation easy

How to Build a Simple Binary Protocol Parser Mock in Python

Defines a mock binary protocol with field definitions, encoding, and decoding to simulate network packet parsing for A/B testing and experiment setup.

binary protocol mock
Python
class SimpleProtocol:
    def __init__(self, name, version):
        self.name = name
        self.version = version
        self.fields = []

    def add_field(self, field_name, field_size):
        self.fields.append((field_name, field_size))

    def parse(self, data):
        if len(data) != sum(size for _, size i…
12 0 Open
A/B testing & experimentation easy

How to Calculate Minimum Sample Size for a T-Test in Python

Compute the minimum sample size per group for a two-sample t-test using effect size, significance level, and statistical power.

sample-size statistics ab-testing
Python
import math
from scipy.stats import norm


def min_sample_size(effect_size, alpha=0.05, power=0.8):
    """
    Calculate minimum sample size for a two-sample t-test (equal groups).

    Args:
        effect_size: Cohen's d (standardized mean difference)
        alpha: significance level (Type I error)
        power: …
15 0 Open
A/B testing & experimentation easy

How to Hash a User ID to an Experiment Bucket in Python

Deterministically map a user ID to one of N experiment buckets using MD5 hashing and modulo arithmetic.

hashing ab-testing bucketing
Python
import hashlib

def hash_to_bucket(user_id: str, num_buckets: int = 10) -> int:
    """Deterministically map a user_id to a bucket (0 to num_buckets-1)."""
    digest = hashlib.md5(user_id.encode("utf-8")).hexdigest()
    return int(digest[:8], 16) % num_buckets

if __name__ == "__main__":
    # Mock experiment: split…
14 0 Open
A/B testing & experimentation easy

How to Mock Stratified Assignment by Segment in Python

Simulate stratified assignment for A/B experiments by sampling a fixed proportion of units from each segment, with deterministic seeds for reproducibility.

ab-testing sampling random
Python
import random

def stratified_assignment(segments, seed=None):
    """
    Mock stratified assignment: given a dict of segment -> population size,
    return a dict of segment -> sampled unit ids (deterministic with seed).
    """
    if seed is not None:
        random.seed(seed)
    rng = random.Random(seed)
    res…
12 0 Open
A/B testing & experimentation medium

How to Perform Intent-to-Treat Analysis in Python

Runs an intent-to-treat analysis on mock A/B test data, comparing outcomes by initial group assignment with a t-test for significance.

ab-testing intent-to-treat statistics
Python
import pandas as pd
import numpy as np


def intent_to_treat_analysis(data):
    """Perform intent-to-treat (ITT) analysis.

    ITT compares outcomes based on initial treatment assignment,
    regardless of whether participants actually received the treatment.
    """
    # Create a copy to avoid mutating the origina…
13 0 Open

Browse by section

Each section groups closely related Python snippets.

Guide: free Python code samples library

Copy-ready Python snippets for learners and developers

PythonSkillset code samples are short, focused examples organised by topic and difficulty. Every snippet is server-rendered HTML — readable by search engines and easy to copy. Open any sample, read the notes, copy the code, then press Try in editor to run it in the browser with Pyodide.

How to use this library

  1. Pick a topic section — strings, lists, files, functions, and more
  2. Open a sample, read How it works, and copy the code block
  3. Run it in the IDE, tweak values, then take a related quiz or tutorial lesson

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.