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 Anti-Corruption Layer in Python
Translate messy legacy system data into a clean domain model using an anti-corruption layer in 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…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 Service Call Timeouts in Python
Simulate service calls with configurable timeouts using Mock to patch sleep and randomness, covering success and timeout cases.
import time
from unittest.mock import Mock, patch
# Simulate a service call with configurable timeout
def call_service(service_name, timeout=5):
"""Mock a service call that may time out."""
start = time.time()
print(f"Calling {service_name}...")
# Simulate service latency (randomized for realism)…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 Mock mTLS Between Services in Python
Simulate mutual TLS authentication between two services using Python's ssl module with self-signed certificates.
import ssl
import socket
import threading
import tempfile
from pathlib import Path
import subprocess
def create_test_cert(cert_path: Path, key_path: Path, common_name: str = "localhost"):
"""Generate a self-signed certificate using openssl."""
subprocess.run([
"openssl", "req", "-x509", "-newkey", "rs…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 …How to implement the Database per service pattern in Python
Simulate separate databases per microservice in Python using dataclasses and in-memory dictionaries, showing how services own their data independently.
import json
from dataclasses import dataclass, asdict
from typing import Dict, List
@dataclass
class User:
id: int
name: str
email: str
@dataclass
class Order:
id: int
user_id: int
product: str
amount: float
class UserServiceDB:
"""Simulates a separate database for the User servic…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…Zero Trust Service Auth Mock in Python
A simple HMAC-based token issuance and validation mock that enforces zero trust between microservices.
import hmac
import hashlib
import json
import time
class ZeroTrustAuth:
def __init__(self, secret_key):
self.secret_key = secret_key
self.service_tokens = {}
def issue_token(self, service_name, ttl=300):
payload = {
"service": service_name,
"issued_at": int(tim…Accumulators Global Counter Mock in Python
Shows an accumulator-style global counter with a mock patch to control its value in tests.
import unittest
from unittest.mock import patch
# Module-level global counter accumulator
counter = 0
def increment(by=1):
"""Increment the global counter in place (accumulator pattern)."""
global counter
counter += by
return counter
def reset():
"""Reset the counter to zero."""
global count…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…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.