Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
How to mock Server-Sent Events (SSE) in Python
A minimal HTTP server that streams Server-Sent Events to clients, perfect for testing and development.
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import time
MESSAGES = iter([
"data: Hello world\n\n",
"data: Second message\n\n",
"event: custom\n",
"data: Custom event payload\n\n",
"data: Final message\n\n"
])
class SSEHandler(BaseHTTPRequestHandler):
def do_GET…Batch Consume Process Commit Pattern in Python
A mock batch processor that accumulates items in a queue, processes full batches, commits successful or failed results, and flushes remaining items.
import random
import threading
import time
from collections import deque
class MockBatchProcessor:
def __init__(self, process_func, commit_func, batch_size=5):
self.queue = deque()
self.batch_size = batch_size
self.process_func = process_func
self.commit_func = commit_func
de…Consistent Hashing Cache Shard in Python
A minimal consistent hashing ring with virtual nodes that distributes cache keys across shards and minimizes re-mapping when a node is removed.
import hashlib
import bisect
class ConsistentHashRing:
def __init__(self, nodes=None, replicas=3):
self.replicas = replicas
self.ring = {}
self.sorted_keys = []
if nodes:
for node in nodes:
self.add_node(node)
def _hash(self, key):
return i…How to Build a Bloom Filter to Reduce Cache Misses in Python
Implement a probabilistic Bloom filter in Python that lets a cache quickly determine which keys are definitely not present, reducing expensive source lookups on cache misses.
import hashlib
import random
class BloomFilter:
def __init__(self, size=100, num_hashes=3):
self.size = size
self.num_hashes = num_hashes
self.bit_array = [0] * size
def _hashes(self, item):
result = []
for i in range(self.num_hashes):
hash_value = int(hash…How to Mock a Redis Transaction with MULTI/EXEC in Python
A minimal in-memory mock of Redis MULTI/EXEC transactions that queues commands and applies them atomically on EXEC.
class RedisTransactionMock:
def __init__(self):
self.data = {}
self.queue = []
self.in_transaction = False
def multi(self):
self.in_transaction = True
self.queue = []
return "OK"
def set(self, key, value):
if self.in_transaction:
self.qu…Implement a TTL cache with a mock clock in Python
This code creates a simple TTL cache that stores values with an expiration timestamp and allows injecting a mock time function to test expiry behavior deterministically.
import time
from functools import wraps
class TTLCache:
def __init__(self, ttl_seconds):
self.ttl = ttl_seconds
self.cache = {}
self._now = time.time
def set_mock_time(self, mock_time_fn):
"""Inject a mock time function for testing TTL expiry."""
self._now = mock_time_…Mock Redis Distributed Lock in Python with SET NX EX
A minimal in-memory mock of Redis SET NX EX distributed lock semantics for testing concurrent code without a real Redis server.
import time
import threading
import uuid
from typing import Optional
class RedisLockMock:
"""A minimal mock of Redis SET NX EX distributed lock semantics."""
def __init__(self):
self._store = {} # key -> (value, expiry_epoch)
def acquire(self, key: str, token: str, ttl_seconds: int) -> bool:
…How to retry idempotent operations with a mock in Python
Wrap a flaky idempotent operation in a retry loop with exponential backoff, and use unittest.mock to deterministically test the str's behavior.
import random
import time
from unittest.mock import Mock
def idempotent_operation(value):
"""Simulate an idempotent operation that sometimes fails."""
if random.random() < 0.6: # 60% failure rate
raise ConnectionError("Temporary failure")
return value * 2
def retry_with_backoff(operation, max_…Leaky Bucket Rate Limiter in Python: Smooth Burst Traffic
Implements a token-bucket-style leaky bucket rate limiter that smooths bursty traffic by draining at a fixed rate and dropping excess packets.
import time
import random
class LeakyBucket:
def __init__(self, capacity, drain_rate):
self.capacity = capacity
self.drain_rate = drain_rate
self.water = 0.0
self.last_time = time.time()
def allow(self, packet_size=1.0):
now = time.time()
elapsed = now - self.…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.
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))
…How to Implement a Two-Phase Commit Mock in Python
Simulate a distributed two-phase commit with prepare, commit, and abort phases, including deterministic failure injection for testing.
import random
from dataclasses import dataclass
from typing import Dict, List, Optional
@dataclass
class Transaction:
tx_id: int
data: Dict[str, str]
class TwoPhaseCommitMock:
"""Simple two-phase commit mock with prepare and commit phases."""
def __init__(self) -> None:
self.prepared: List…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.
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…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.
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.…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.
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 …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.
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])
…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…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…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.
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…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.
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…Consistent Hashing with Virtual Buckets in Python
This code maps many virtual buckets onto a few physical buckets using a consistent hashing ring, ensuring balanced distribution with minimal remapping when physical buckets change.
import random
class VirtualBuckets:
"""Maps many virtual buckets onto few physical buckets using consistent hashing."""
def __init__(self, physical_buckets, virtual_factor=100):
self.physical = list(physical_buckets)
self.virtual_factor = virtual_factor
self.ring = []
self…How to Implement Consistent Hashing in Python
Build a consistent hash ring in Python that distributes keys across nodes and minimizes remapping when nodes are added or removed.
import hashlib
from bisect import bisect_right
class ConsistentHashRing:
def __init__(self, nodes, replicas=3):
self.replicas = replicas
self.ring = {}
self.sorted_keys = []
for node in nodes:
self.add_node(node)
def _hash(self, key):
return int(hashlib.md…ACME LetsEncrypt Mock Challenge Server in Python
A minimal HTTP server that serves key authorizations for ACME/Let's Encrypt DNS-01 or HTTP-01 challenges during testing and validation.
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
# In-memory store simulating the ACME challenge token -> key authorization pair
challenge_store = {
"token_example": "token_example.key_authorization"
}
class AcmeChallengeHandler(BaseHTTPRequestHandler):
def do_GET(self):
# Extra…ChaCha20-Poly1305 mock in Python
Simulates ChaCha20-Poly1305 AEAD encryption and authentication using SHA-256 as a deterministic keystream and tag generator.
from hashlib import sha256
import struct
def chacha20_block(key, counter, nonce):
"""Mock ChaCha20 block: deterministic pseudo-random keystream from key+counter+nonce."""
state_input = key + struct.pack("<I", counter) + nonce + b"ChaCha20"
return sha256(state_input).digest()[:64] # 64-byte keystream bloc…How to Mock KMS Envelope Encryption in Python
Demonstrates a minimal mock of AWS KMS envelope encryption flow with AES-GCM data key wrapping and unwrapping.
import base64
import json
import os
import hashlib
class MockKMS:
"""Minimal mock of AWS KMS envelope encryption flow."""
def generate_data_key(self):
# Simulate KMS returning a plaintext and encrypted data key
plaintext_key = os.urandom(32)
encrypted_key = hashlib.sha256(plaintext_k…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
- Pick a topic section — strings, lists, files, functions, and more
- Open a sample, read How it works, and copy the code block
- 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.