Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
Strangler Fig Migration Pattern in Python
Gradually reroute calls from a legacy service to a modern replacement using a runtime switch and feature detection.
from dataclasses import dataclass
@dataclass
class PaymentService:
def process(self, amount: float) -> str:
return f"Legacy processed ${amount:.2f}"
class StranglerFig:
def __init__(self):
self._new_service = None
def attach_new(self, service):
self._new_service = service
de…Compaction Small Files Mock in Python
Simulates a small-files compaction job by creating small mock files and merging them into a single output file using Python's standard library.
from pathlib import Path
import tempfile
import os
def create_small_files(directory: Path, file_count: int = 5, lines_per_file: int = 3):
"""Create several small mock files with sample content."""
directory.mkdir(exist_ok=True)
for i in range(file_count):
file_path = directory / f"part-{i:04d}.tx…How to Create a Mock Kafka Producer in Python
Build a Kafka producer that generates mock streaming records with JSON serialization and error handling for local testing.
import json
import time
from kafka import KafkaProducer
from kafka.errors import KafkaError
def create_mock_producer(bootstrap_servers="localhost:9092", topic="input-topic"):
"""Create a Kafka producer that generates mock streaming data."""
producer = KafkaProducer(
bootstrap_servers=bootstrap_servers…How to Filter and Project Spark DataFrames with PySpark SQL
Simulate a SQL SELECT with WHERE using PySpark DataFrame select and filter to project columns and apply conditions.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("QueryFilterMock").master("local[2]").getOrCreate()
data = [
("Alice", 28, "Engineering"),
("Bob", 35, "Sales"),
("Carol", 32, "Engineering"),
("David", 25, "Marketing"),
("Eve", 29, "E…How to Implement MapReduce Word Count in Python Using a Dict
Simulate a MapReduce word count pipeline in Python with a mock dict, splitting text into words, shuffling, and reducing to frequency counts.
def map_reduce_word_count(text: str) -> dict:
"""Simulate a MapReduce pipeline to count word frequencies."""
# MAP phase: split into words and emit (word, 1) pairs
mapped = []
for word in text.lower().split():
# Clean word of punctuation
clean_word = ''.join(char for char in word if cha…How to Implement collect_list in Python
Group rows by a key and collect all corresponding values into a list — a pure-Python mock of Spark's collect_list aggregation.
from collections import defaultdict
def collect_list(rows, key_field, value_field):
grouped = defaultdict(list)
for row in rows:
grouped[row[key_field]].append(row[value_field])
return dict(grouped)
if __name__ == "__main__":
data = [
{"dept": "sales", "emp": "alice"},
{"dept"…How to Mock DataFrame Schema Columns in Python
Create an empty pandas DataFrame with only the specified column names to mock a schema before any data is loaded.
import pandas as pd
def mock_schema(columns):
return pd.DataFrame(columns=columns)
if __name__ == "__main__":
cols = ["name", "age", "city"]
df = mock_schema(cols)
print(df)
print(f"Columns: {list(df.columns)}, Shape: {df.shape}")How to Mock Hive Support in PySpark with unittest.mock
This code demonstrates how to mock Hive support in a PySpark environment using unittest.mock to simulate SQL queries returning fixed data.
from unittest.mock import Mock, patch
def get_hive_tables(spark):
"""Mock Hive support by returning a fixed list of tables."""
return spark.sql("SHOW TABLES").collect()
class HiveTable:
"""Simple class that mimics a Hive table row."""
def __init__(self, database, tableName):
self.database =…How to Mock a File Source Watch Directory in Python
Poll a directory for new files and log changes, simulating a watch directory for data ingestion patterns.
import os
import time
from pathlib import Path
def watch_directory(dir_path: str, poll_interval: float = 1.0, max_iterations: int = 5):
"""
Mock a file-source watch directory by polling for changes.
Returns new files detected during each poll cycle.
"""
directory = Path(dir_path)
directory.mk…How to Mock a Socket Stream in Python
Simulate a streaming socket source with a generator to test stream-read and buffering logic without a real network.
import socket
import threading
import time
def mock_socket_stream(data_chunks, delay=0.1):
"""Generator that simulates a streaming socket source."""
for chunk in data_chunks:
time.sleep(delay)
yield chunk
def read_stream_socket(stream_gen):
"""Reads from mock stream and prints received ch…How to Mock a User-Defined Function (UDF) in Python
Wrap a real UDF implementation with call logging to simulate and track invocations in a data pipeline.
from typing import Any, Callable
# Mock a user-defined function (UDF) that was previously complex or external
def mock_udf(name: str, implementation: Callable[..., Any], *, calls: list[Any]) -> Callable[..., Any]:
"""Wrap a real implementation with call logging to simulate a UDF."""
def wrapper(*args: Any, *…How to Pivot and Group Aggregate in Python
Group records by a key, collect values, and apply an aggregate function (like sum) to build a pivot-style summary dictionary.
from collections import defaultdict
def pivot_group_aggregate(records, group_key, value_key, agg_func):
groups = defaultdict(list)
for record in records:
groups[record[group_key]].append(record[value_key])
return {key: agg_func(values) for key, values in groups.items()}
if __name__ == "__main__":…How to Shuffle Items by Group in Python
Randomly shuffle items within each group while keeping groups contiguous, using a seed for reproducible results.
import random
def shuffle_sort_groups(items, group_key, seed=None):
"""Randomize order within groups, keeping groups contiguous."""
rng = random.Random(seed)
groups = {}
for item in items:
key = group_key(item)
groups.setdefault(key, []).append(item)
result = []
for k…How to Truncate Lineage Back to a Checkpoint in Python
Walks a linked list of lineage nodes upward to find the nearest checkpoint and returns that node, truncating the lineage.
class LineageNode:
def __init__(self, name, parent=None, checkpoint=None):
self.name = name
self.parent = parent
self.checkpoint = checkpoint
def truncate_at_checkpoint(self):
"""Truncate lineage back to the last checkpoint."""
current = self
while current.check…How to Use Broadcast Variables as Read-Only in PySpark (Mock Example)
Share a lookup dict across Spark executors with a broadcast variable and verify its read-only behavior in a local mock.
from pyspark import SparkContext, SparkConf
def main():
conf = SparkConf().setAppName("BroadcastMock").setMaster("local[2]")
sc = SparkContext(conf=conf)
lookup = {"a": 1, "b": 2, "c": 3}
broadcast_lookup = sc.broadcast(lookup)
data = ["a", "b", "c", "a", "unknown"]
rdd = sc.parallel…How to select specific columns in Python with SQLite
A reusable function that connects to a SQLite database and returns only the requested columns from a given table.
import sqlite3
def select_pruned_columns(db_path, table, columns):
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
col_list = ", ".join(columns)
query = f"SELECT {col_list} FROM {table}"
return cursor.execute(query).fetchall()
if __name__ == "__main__":
conn = sq…Hudi Upsert Mock Copy on Write in Python
Simulates Apache Hudi's Copy-on-Write upsert behavior by merging update records into a deep copy of base records, replacing matches or appending new ones.
import copy
from typing import Dict, List, Any
def upsert_copy_on_write(base_records: List[Dict[str, Any]], updates: List[Dict[str, Any]], key_field: str = "id") -> List[Dict[str, Any]]:
"""Simulate Hudi Copy-on-Write upsert: merge updates into a copy of base records."""
result = copy.deepcopy(base_records)
…Modeling a Hive Metastore Table Schema in Python
A dataclass that mimics a Hive metastore table schema—columns, partition keys, storage format, and location—with helper methods for description and mutation.
from dataclasses import dataclass, field
from typing import Dict, List, Optional
@dataclass
class HiveTable:
"""Simple mock of a Hive metastore table schema."""
name: str
database: str = "default"
columns: List[Dict[str, str]] = field(default_factory=list)
partition_keys: List[Dict[str, str]] = f…Partition Data by Hash Key Mod N in Python
Returns a partition index for a string key by hashing it with MD5 and taking modulo N, then groups sample keys into partitions.
import hashlib
def partition_key(key: str, num_partitions: int) -> int:
"""Return partition index for key using MD5 hash mod N."""
digest = hashlib.md5(key.encode()).hexdigest()
return int(digest, 16) % num_partitions
if __name__ == "__main__":
keys = ["alice", "bob", "carol", "dave", "eve"]
nu…Session window gap mock in Python
Group sorted timestamps into sessions where any gap between consecutive events exceeds a threshold starts a new session.
from datetime import datetime, timedelta
def session_windows(timestamps, gap_seconds=300):
"""Group timestamps into sessions where gaps > gap_seconds start new sessions."""
if not timestamps:
return []
# Sort timestamps chronologically to ensure correct windowing
timestamps = sorted(timestam…Sliding Window Streaming Mock in Python
A simple Python class that maintains a sliding window of recent streaming values and computes the running average.
import time
import random
class StreamingMock:
"""Produces a stream of numbers using a sliding window."""
def __init__(self, window_size=5):
self.window = []
self.window_size = window_size
def push(self, value):
"""Add a value, sliding the window forward."""
s…Z-Order Optimization in Python
A mock concept demonstrating z-order layout optimization by reassigning z-indices based on areas size.
class ZOrderLayout:
"""
Minimal mock for z-order layout optimization using a stacking score.
Elements overlap; higher z_index is drawn on top.
"""
def __init__(self):
self.elements = []
def add_element(self, name, area, z_index):
self.elements.append({"name": name, "area": area…Build a Mock Random Forest Classifier in Python
Create a simple random-forest-like classifier with random majority voting between trees, including fit, predict, and predict_proba methods.
import random
class MockRandomForest:
def __init__(self, n_trees=10, random_state=42):
self.n_trees = n_trees
self.random_state = random_state
self.classes_ = None
self._class_counts = None
random.seed(random_state)
def fit(self, X, y):
self.classes_ = sorted(…Compare Model A vs Model B Metrics in Python
A script that simulates and compares metrics between two ML models, showing a formatted diff table for quick insight.
import random
def compare_a_b(samples=5):
"""Mock comparison of model A vs model B predictions."""
metrics = ["accuracy", "precision", "recall", "f1"]
print(f"{'Metric':<12}{'Model A':>10}{'Model B':>10}{'Diff':>10}")
print("-" * 42)
random.seed(42)
for metric in metrics:
a = round(r…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.