Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
Fallback cached response mock in Python
Wraps a mock function with a fallback to a real service and caches results to mask transient failures.
import time
from functools import wraps
class CachedMock:
def __init__(self, cache_ttl=5):
self.cache = {}
self.cache_ttl = cache_ttl
def get(self, key):
cached = self.cache.get(key)
if cached and time.time() - cached["timestamp"] < self.cache_ttl:
return cached["v…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 Handle mTLS Certificate Rotation in Python
Detect mTLS certificate file changes by tracking modification time and hot-reload the SSL context in a running service.
import ssl
import tempfile
import datetime
from pathlib import Path
class MTLSContext:
def __init__(self, cert_path, key_path, ca_path):
self.cert_path = Path(cert_path)
self.key_path = Path(key_path)
self.ca_path = Path(ca_path)
self.context = None
self.last_loaded_mtime …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 Choreography Saga in Python
Simulate a choreography-based saga with event envelopes, status tracking, and compensating actions to model distributed transactions.
import json
from dataclasses import dataclass, asdict
from typing import List, Optional
from enum import Enum
class SagaStatus(Enum):
PENDING = "PENDING"
COMPLETING = "COMPLETING"
COMPLETED = "COMPLETED"
FAILED = "FAILED"
@dataclass
class EventEnvelope:
event_type: str
order_id: str
sta…How to implement a circuit breaker in Python
A Python CircuitBreaker class that tracks failures, opens after a threshold, and retries after a timeout.
class CircuitBreaker:
def __init__(self, failure_threshold=3, timeout=5):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED"
def call(self, mock_downstream):
if self.state …JWT Service-to-Service Authentication Mock in Python
Create and verify HS256 JWTs for service-to-service authentication without external libraries.
import hashlib
import hmac
import base64
import json
import time
class JWTMock:
"""Minimal JWT service-to-service mock using HS256."""
def __init__(self, secret):
self.secret = secret.encode()
@staticmethod
def _b64url_encode(data):
return base64.urlsafe_b64encode(data).rstr…Python Saga Compensating Steps Mock
Mock a distributed transaction saga with forward steps and compensating actions that reverse partial progress on failure.
from datetime import datetime
def make_payment(user_id, amount):
print(f"[{datetime.now():%H:%M:%S}] Payment of ${amount} processed for user {user_id}")
return {"step": "payment", "status": "ok", "details": f"${amount} charged"}
def deduct_inventory(order_id, items):
print(f"[{datetime.now():%H:%M:%S}]…Saga pattern orchestration with rollback in Python
Orchestrate a distributed transaction with Saga steps and automated compensation rollback on failure.
import time
import random
class SagaStep:
def __init__(self, name):
self.name = name
self.executed = False
def execute(self):
print(f"Executing {self.name}...")
time.sleep(0.2)
if random.random() < 0.3:
raise RuntimeError(f"{self.name} failed")
sel…Approximate Distinct Count in Python with HyperLogLog
Mock a large data stream and estimate the number of distinct items with a HyperLogLog-style probabilistic counter to save memory.
import random
import string
from collections import Counter
import math
class ApproxCountDistinct:
def __init__(self, num_buckets=16):
self.num_buckets = num_buckets
self.max_zeros = [0] * num_buckets
def _hash(self, item):
# Simple string hash to a 32-bit integer
h = …Bloom Filter Join Mock in Python
A mock hash join that uses a Bloom filter to pre-filter one table before performing an exact match, reducing the number of comparisons in large dataset joins.
import hashlib
import random
import string
class BloomFilter:
def __init__(self, size: int = 200, num_hashes: int = 3):
self.bits = [False] * size
self.size = size
self.num_hashes = num_hashes
def _hashes(self, item: str):
result = []
for seed in range(self.num_hashes…Delta Lake ACID Transaction Log Mock in Python
Simulates Delta Lake's transactional log with JSON files for atomic commits, versioned operations, and crash recovery
import json
import time
from pathlib import Path
class DeltaLog:
def __init__(self, path):
self.log_dir = Path(path)
self.log_dir.mkdir(parents=True, exist_ok=True)
self.version = 0
def _write_txn(self, action, payload):
txn = {
"version": self.version,
…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.
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…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.
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"…How to Implement a Mock MapReduce for Word Count in Python
Simulates a MapReduce word count pipeline with mapper, shuffle, and reducer phases using Python dicts and standard library modules.
from collections import defaultdict
import re
def mapper(text):
"""Split text into words and emit (word, 1) pairs."""
words = re.findall(r'\b\w+\b', text.lower())
return [(word, 1) for word in words]
def reducer(pairs):
"""Group word-count pairs and sum counts."""
counts = defaultdict(int)
fo…How to Implement a Streaming Watermark in Python
Mock structured streaming watermarks in Python to track late event times and compute a watermark for windowed processing.
from datetime import datetime, timedelta
import time
class StreamingWatermark:
"""Mock watermark tracker for structured streaming."""
def __init__(self, watermark_delay_seconds):
self.watermark_delay = timedelta(seconds=watermark_delay_seconds)
self.max_event_time = None
def observe_even…How to Implement row_number Window Function in Python
This code implements a SQL-style ROW_NUMBER() window function in pure Python, partitioning rows by a set of columns and ranking them within each partition by an ordered set of columns.
from collections import defaultdict
import itertools
def row_number(rows, partition_by, order_by):
partitions = defaultdict(list)
for index, row in enumerate(rows):
key = tuple(row[col] for col in partition_by)
partitions[key].append((index, row))
result = []
for key in partitions:
…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…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.
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…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…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.
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…How to Simulate a MapReduce Mock with Combine Phase in Python
Simulates a MapReduce pipeline with a combiner that aggregates local counts per reducer to reduce network and compute overhead.
from collections import defaultdict
def map_phase(lines):
intermediate = defaultdict(list)
for line in lines:
for word in line.strip().lower().split():
intermediate[word].append(1)
return dict(intermediate)
def combine_phase(intermediate, num_reducers=3):
combined = defaultdict(li…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.
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:
…How to use foreachBatch with a mock sink in PySpark
Demonstrates using Spark Structured Streaming's foreachBatch sink to capture and verify streaming batches by writing them into a custom mock sink object.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit
class MockSink:
def __init__(self):
self.batches = []
def write_batch(self, batch_df, batch_id):
# Collect batch data as list of dicts for verification
records = batch_df.collect()
self.batches…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.