Reference library

Python Code Samples

Medium snippets you can copy, study, and run in the browser editor.

102 matches
Observability & SRE medium

How to Build a Burn Rate Alert with Multiple Time Windows in Python

Track token consumption and trigger alerts when the burn rate exceeds a threshold across multiple time windows using deque and time-based sliding windows.

burn-rate alerts time-windows
Python
import time
from collections import deque

class BurnRateAlert:
    def __init__(self, windows_seconds=(60, 300, 900), threshold_rate=0.8):
        self.windows = {w: deque() for w in windows_seconds}
        self.threshold_rate = threshold_rate
        self.previous_tokens = None

    def record_sample(self, current_…
16 0 Open
Observability & SRE medium

How to Build a Python Latency Histogram with Mock Buckets

This code implements a mock latency histogram that records request durations into configurable buckets and outputs counts, total, and average latency.

histogram latency metrics
Python
import time
import random
from collections import Counter


class LatencyHistogram:
    def __init__(self, buckets):
        self.buckets = sorted(buckets)
        self.counts = Counter()
        self.total = 0
        self.sum_latency = 0

    def record(self, latency_ms):
        for i, boundary in enumerate(self.bu…
13 0 Open
Observability & SRE medium

How to Build an HTTP Server Request Duration Histogram in Python

Create a small HTTP server that times each GET request, buckets the duration, and prints a histogram on shutdown.

http.server histogram performance
Python
import time
import random
from collections import Counter
from http.server import HTTPServer, BaseHTTPRequestHandler


class HistogramHandler(BaseHTTPRequestHandler):
    response_times = Counter()

    def do_GET(self):
        start = time.perf_counter()
        time.sleep(random.uniform(0.001, 0.1))
        duratio…
13 0 Open
Observability & SRE medium

Summary Quantile Mock Sketch in Python

Build a memory-efficient sketch that stores sorted bins of data points to answer approximate quantile queries like median without keeping all values in memory.

quantile sketch statistics
Python
import random
import statistics
from collections import Counter

class SummaryQuantileSketch:
    """
    A simple sketch that stores a fixed-size summary of data (min, max, deciles)
    using sorted bins, then answers approximate quantile queries.
    """
    def __init__(self, bins=10):
        self.bins = bins
    …
13 0 Open
Microservices patterns medium

How to Build an Anti-Corruption Layer in Python

Translate messy legacy system data into a clean domain model using an anti-corruption layer in Python.

anti-corruption microservices data-transformation
Python
class MockLegacySystem:
    """Simulates a legacy system with messy data formats."""
    def get_user_data(self):
        # Legacy format: fields are abbreviated and types are inconsistent
        return {
            "usr_id": "USR-123",
            "usr_nm": "john_doe",
            "email_addrs": "John.Doe@example.c…
15 0 Open
Microservices patterns medium

How to Build an OAuth Client Credentials Mock Server in Python

A minimal HTTP mock server implementing the OAuth 2.0 client credentials grant for local testing and microservice development.

oauth mock-server microservices
Python
from http.server import HTTPServer, BaseHTTPRequestHandler
import json

TOKENS = {"valid_token": "demo_access_token", "client_id": "my_service"}

class OAuthHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path == "/oauth/token":
            length = int(self.headers.get("Content-Length", 0))
  …
14 0 Open
Big data & Spark medium

How to Build a DAG Execution Stage Calculator in Python

Computes the execution stages of a directed acyclic graph (DAG) by grouping nodes that become ready simultaneously using topological sorting with Kahn's algorithm.

dag topological-sort kahn-algorithm
Python
from collections import defaultdict, deque


def get_stages(edges):
    """Return list of stages, where each stage is a list of nodes
    that become ready at the same time in a DAG."""
    graph = defaultdict(list)
    in_degree = defaultdict(int)
    nodes = set()

    for src, dst in edges:
        graph[src].appen…
15 0 Open
Big data & Spark medium

How to Create a Mock Iceberg Snapshot Manifest in Python

Build a mock Iceberg snapshot manifest structure with metadata and data entries using Python dictionaries and JSON.

iceberg manifest snapshot
Python
import json
from datetime import datetime, timezone


def create_mock_manifest(snapshot_id: int, file_paths: list[str]) -> dict:
    """Create a mock Iceberg snapshot manifest structure."""
    manifest_file = {
        "manifest_path": f"/warehouse/table/metadata/snap-{snapshot_id}-m0.avro",
        "manifest_length"…
15 0 Open
Big data & Spark medium

How to Mock a Catalyst Logical Plan in Python

Build a small Python class that mimics Spark Catalyst's logical plan tree for teaching or testing query optimizations.

apache-spark logical-plan catalyst
Python
from typing import Any, Dict, List, Optional


class CatalystLogicalPlan:
    """A minimal mock of Catalyst's logical plan for teaching purposes."""
    
    def __init__(self, node_type: str, **kwargs: Any) -> None:
        self.node_type = node_type
        self.attributes: Dict[str, Any] = kwargs
        self.child…
13 0 Open
Big data & Spark medium

How to Mock and Test a Rate-Limited Source Stream in Python

Build a class that rate-limits emitted items using a sliding window and test it with a simulated stream in Python.

rate-limiting mock-testing streaming
Python
import time
from collections import deque


class RateLimitedSource:
    def __init__(self, max_rate, window=1.0):
        self.max_rate = max_rate
        self.window = window
        self._timestamps = deque()

    def emit(self, item):
        now = time.monotonic()
        while self._timestamps and self._timestam…
16 0 Open
Big data & Spark medium

How to implement a tumbling window aggregation in Python

Build a mock tumbling window aggregator in Python that groups streaming events into fixed time intervals and computes count, sum, and average per window.

tumbling-window streaming aggregation
Python
import time
from collections import deque

class TumblingWindow:
    def __init__(self, duration_seconds):
        self.duration = duration_seconds
        self.buffer = deque()
        self.window_start = None

    def add(self, item):
        current_time = time.time()
        if self.window_start is None:
         …
13 0 Open
Big data & Spark medium

Lazy Evaluation Transform Lineage Mock in Python

Build a mock lineage tracker for data transforms using lazy evaluation and function wrappers in Python.

lazy-evaluation lineage decorator
Python
import functools


def lazy_transform(pipeline):
    """Build a mock lineage tracker using lazy evaluation."""
    lineage = []

    def wrap(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            lineage.append({"transform": func.__name__, "a…
16 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 medium

How to Build an sklearn Pipeline with ColumnTransformer in Python

A mock example showing how to chain preprocessing and a regression model into a single sklearn Pipeline, scaling numeric features and one-hot encoding categorical features with ColumnTransformer.

sklearn pipeline columntransformer
Python
import numpy as np
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LinearRegression

# Mock dataset
X = np.array([[1, 'red'], [2, 'blue'], [3, 'red'], [4, 'green'], [5, 'blue']], dtype=o…
13 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 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 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 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 medium

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.

flyte ml-pipeline dataclass
Python
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…
16 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

How to Simulate Geo Experiments in Python

Build a mock geo experiment simulator with ramp-up/down periods, measuring weekly lift between treatment and control markets.

geo-experiment ab-testing simulation
Python
import random
import math
from dataclasses import dataclass

@dataclass
class GeoMarket:
    name: str
    base_demand: float
    geo_coefficient: float

def simulate_geo_experiment(markets, weeks=12, control_weeks=6):
    """
    Simulates a geo experiment with ramp-up and ramp-down periods.
    Returns weekly lift p…
18 0 Open
Database scaling & optimization medium

Build a Full Text Search Index in Python

Create a simple inverted index for full-text search with the standard library, supporting multi-word AND queries across documents.

search inverted-index text-processing
Python
import re
from collections import defaultdict


class SimpleTextIndex:
    def __init__(self):
        self.index = defaultdict(list)
        self.documents = {}

    def add_document(self, doc_id, text):
        self.documents[doc_id] = text
        words = set(re.findall(r'\w+', text.lower()))
        for word in wo…
12 0 Open
Database scaling & optimization medium

Database Helper in Python with SQLite Scaling Optimization

Build a beginner-friendly SQLite database helper class with WAL, indexed queries, and efficient batch inserts for scaling.

sqlite database scalability
Python
import sqlite3
from contextlib import contextmanager


class DatabaseHelper:
    """Beginner-friendly helper for SQLite database operations with scaling tips."""

    def __init__(self, db_path):
        self.db_path = db_path

    @contextmanager
    def connection(self):
        """Context manager for automatic comm…
14 0 Open
Database scaling & optimization medium

How to Build a Connection Pool Reuse Mock in Python

Build a mock connection pool with context manager to track connection reuse, acquires, and releases in Python.

connection-pool context-manager database
Python
import time
from contextlib import contextmanager


class Connection:
    def __init__(self, name):
        self.name = name
        self.in_use = False
        self.busy_since = None

    def fetch(self):
        return f"data from {self.name}"


class ConnectionPool:
    def __init__(self, size=3):
        self.conn…
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.