Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
Chunk Large File Upload Simulation by Blocks in Python
A Python script reads a large binary file in fixed-size chunks and simulates a block-by-block upload with per-chunk SHA256 hashing.
import os
import hashlib
from pathlib import Path
def read_file_in_chunks(file_path, chunk_size=8196):
"""Yield chunks of a file as bytes."""
with open(file_path, 'rb') as f:
while chunk := f.read(chunk_size):
yield chunk
def simulate_chunked_upload(file_path, chunk_size=8196):
"""S…Binary Search for Ship Capacity in Python
Use binary search to find the minimum ship capacity that can transport all packages within a given number of days.
def ship_within_days(weights, days):
def can_ship(capacity):
current = 0
needed_days = 1
for weight in weights:
if current + weight > capacity:
needed_days += 1
current = 0
current += weight
return needed_days <= days
low …Game of Life Next State Grid in Python
Compute the next generation of Conway's Game of Life from a 2D grid using the standard three rules with neighbor counting.
def next_state(grid):
m, n = len(grid), len(grid[0])
new = [[0] * n for _ in range(m)]
for r in range(m):
for c in range(n):
total = 0
for dr in (-1, 0, 1):
for dc in (-1, 0, 1):
if dr == 0 and dc == 0:
continue
…How to Implement a Weighted DNS Resolver with Failover in Python
Simulates a weighted DNS load balancer that distributes traffic across IPs by weight and automatically fails over when a server is marked unhealthy.
import random
import time
class WeightedDNSResolver:
def __init__(self, records):
self.records = records # list of (ip, weight)
self.total_weight = sum(weight for _, weight in records)
self.failed_ips = set()
def resolve(self):
available = [(ip, weight) for ip, weight in self…Simulate a Leaky Bucket Rate Limiter in Python
This code implements a leaky bucket rate limiter that drains at a fixed rate and accepts or rejects incoming requests based on capacity.
import time
from collections import deque
class LeakyBucket:
"""Simulates a leaky bucket rate limiter with a fixed drain rate."""
def __init__(self, capacity, drain_rate_per_sec):
self.capacity = capacity
self.drain_rate = drain_rate_per_sec
self.water = 0.0
self.last_refill =…How to Simulate RabbitMQ Exchange Routing in Python
Simulate RabbitMQ exchange routing using a nested dict, matching routing keys against patterns like error.* and info.# to return bound queues.
from collections import defaultdict
def route_message(exchanges, exchange_name, routing_key):
"""
Simulate RabbitMQ exchange routing using a nested dict structure.
Returns list of queue names that match the routing key.
"""
queues = exchanges.get(exchange_name, {})
matched = []
for pa…Implement a retry queue with visibility timeout in Python
This code simulates a message queue with a visibility timeout, allowing messages to be retried if not deleted before the timeout expires.
import time
from collections import deque
class SimpleQueue:
def __init__(self, visibility_timeout=2):
self.queue = deque()
self.in_flight = {}
self.visibility_timeout = visibility_timeout
def send(self, message):
self.queue.append(message)
def receive(self):
if …Mock Watermark Late Event Side Output in Python
Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.
from datetime import datetime, timedelta
from typing import List, Tuple
def watermark_mock(
events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
"""Simulate watermarking: events arriving on time vs. late by ch…How to Mock a Liveness Check and Restart a Process in Python
Simulate a failing process and restart it after a liveness check fails, using a mock class and a liveness loop.
import subprocess
import sys
import time
import os
class ProcessMock:
def __init__(self, name, fail_after_seconds=3):
self.name = name
self.fail_after = fail_after_seconds
self.start_time = None
self.is_running = False
def start(self):
self.start_time = time.time()
…How to Send Messages to a Dead Letter Queue in Python
Simulates a poison message queue that retries failed messages up to a limit before moving them to a dead letter queue.
import json
class PoisonMessageQueue:
def __init__(self, max_retries=3):
self.dlq = []
self.max_retries = max_retries
self.processed_count = 0
self.failed_count = 0
def process_message(self, message_body):
if "poison" in message_body:
self.failed_count += 1…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.…Mock a Two-Phase Commit Coordinator in Python
Simulates a two-phase commit protocol where a coordinator asks participants to prepare, then commits or aborts based on unanimous readiness.
import random
import time
from typing import Dict, List
class TwoPhaseCommitCoordinator:
def __init__(self, participants: List[str]):
self.participants = participants
self.participant_state: Dict[str, bool] = {}
def prepare(self) -> bool:
print("[Coordinator] Phase 1: Prepare")
…How to Mock Spark Streaming Micro-Batches in Python
Simulate Spark's micro-batch streaming with a simple deque-based class that collects events over time and processes them in timed batches.
import time
from collections import deque
from datetime import datetime
class MicroBatchStream:
def __init__(self, batch_interval_sec=2):
self.batch_interval = batch_interval_sec
self.source = deque()
self.processed = []
def add_events(self, events):
self.source.extend(events…Mock RDD in Python: Simulate Spark RDD Lazy Transformations
Simulate Apache Spark RDD behavior in Python with lazy maps, filters, partitions, and a collect action.
import random
def mock_rdd(data, num_slices=2):
"""
A simple simulation of Spark RDD behavior with lazy evaluation,
transformations, and an action.
"""
class SimpleRDD:
def __init__(self, data, num_slices=2):
self.data = data
self.num_slices = num_slices
…Check Sample Ratio Mismatch in Python
Estimates the probability that a simple random sample's proportion differs from the population proportion by more than 10% using simulation.
import random
def sample_ratio_mismatch(population_size: int, sample_size: int, p: float) -> float:
"""
Estimate the probability that a simple random sample's proportion
differs from the population proportion by more than 10%.
"""
total_counts = [0, 0]
for _ in range(10000):
sample = …Epsilon Greedy Bandit Mock in Python
A simple epsilon-greedy multi-armed bandit simulation that balances exploration and exploitation to estimate true means of several Bernoulli-like reward distributions.
import random
class Bandit:
def __init__(self, true_mean):
self.true_mean = true_mean
self.estimated_mean = 0.0
self.n_pulls = 0
def pull(self):
return random.gauss(self.true_mean, 1.0)
def update(self, reward):
self.n_pulls += 1
self.estimated_mean += (r…How to Create an Interrupted Time Series Mock in Python
Generate simulated interrupted time series data with a pre/post-intervention trend, level shift, and noise to test segmented regression models.
import numpy as np
# Mock interrupted time series data
np.random.seed(42)
n_pre = 50
n_post = 50
time = np.arange(0, n_pre + n_post)
# Pre-intervention: linear trend + noise
pre_trend = 0.05 * time[:n_pre] + np.random.normal(0, 0.5, n_pre)
# Post-intervention: new slope + level shift + noise
post_trend = 0.05 * tim…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.
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…How to simulate a contextual bandit in Python
Simulate a contextual multi-armed bandit with random features and epsilon-greedy action selection in Python.
import random
class ContextualBandit:
def __init__(self, n_actions=3, n_features=4):
self.n_actions = n_actions
self.n_features = n_features
self.theta = [random.random() for _ in range(n_actions * n_features)]
def mock_context(self):
return [random.uniform(-1, 1) for _ in ra…Cross Shard Query Scatter Gather Mock in Python
Simulate a distributed database cross-shard query using a scatter-gather pattern with a mock Python implementation.
from dataclasses import dataclass
from typing import List, Dict
@dataclass
class NodeResponse:
node_id: int
data: Dict[str, float]
def mock_query_shard(shard_id: int, shard_data: Dict[str, float], query: str) -> NodeResponse:
"""Simulate querying a single shard, returning matches whose value > 50."""
…How to Simulate Colocated Shard Joins in Python
Groups shards by their node and merges co-located shards into a single logical unit, checking capacity constraints.
import random
from collections import defaultdict
def simulate_colocated_shards_join(nodes: list[dict], shards: list[dict]) -> dict:
"""
Simulates the join of co-located shards (on the same node) into a single
logical shard. Returns the resulting node-to-shard mapping.
Each node: {'id': str, 'capaci…How to Simulate Distributed Transactions in Python with a Mock
Model distributed transaction behavior with a mock Transaction class that supports commit, rollback, and failure simulation.
class Transaction:
def __init__(self, id):
self.id = id
self.operations = []
self.committed = False
def add_operation(self, op, data):
self.operations.append((op, data))
def commit(self):
if not self.operations:
raise ValueError("No operations to commit…Auto Rollback on Error Rate Exceeded in Python
Simulate a service that monitors a rolling window of request errors and automatically rolls back when the error rate exceeds a threshold.
import random
import time
def simulate_requests(total_requests=1000, rollback_threshold=0.2):
"""
Simulate a service that automatically rolls back when the error rate
exceeds a threshold within a rolling window.
"""
window_size = 100
errors_seen = []
rolled_back = False
for req_num i…How to Mock Canary Deployment Traffic Split in Python
Simulate a canary deployment's stable/canary traffic split using deterministic request hashing to mock rollout behavior with precise percentage control.
class CanaryDeployment:
def __init__(self, stable_weight: float = 0.9, canary_weight: float = 0.1):
self.stable_weight = stable_weight
self.canary_weight = canary_weight
self.total_weight = stable_weight + canary_weight
def route_request(self, request_id: int) -> str:
"""Route …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.