Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Mock Kafka Topic Partitions with a Python dict of lists
Mocks a Kafka topic and its partitions using a defaultdict of lists to simulate message production, consumption, and per-partition counts.
from collections import defaultdict
class KafkaTopicPartitionMock:
"""A simple mock for Kafka topic-partition assignment using dict of lists."""
def __init__(self, topic):
self.topic = topic
self.partitions = defaultdict(list) # partition_id -> list of messages
def produce(self, message…How to Mock a Kafka Rebalance Listener in Python
Simulate Kafka consumer rebalance callbacks (on_partitions_revoked and on_partitions_assigned) with a mock consumer to test listener logic.
import time
from collections import defaultdict
class MockKafkaConsumer:
def __init__(self):
self.assignments = defaultdict(list)
self.rebalances = 0
def assign(self, partitions):
self.rebalances += 1
self.assignments.clear()
for partition in partitions:
s…Implement the Transactional Outbox Pattern with SQLite in Python
A Python implementation of the transactional outbox pattern using SQLite, ensuring atomic writes of order data and outbox events in a single transaction while supporting reliable message publishing and consumption.
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
import json
@dataclass
class Order:
order_id: str
amount: float
status: str
class TransactionalOutbox:
def __init__(self, db_path=":memory:"):
self.conn = sqlite3.connect(db_path)
self._create_tab…In-Memory PubSub Topic Subscribe Mock in Python
Build a thread-safe in-memory publish/subscribe mock where handlers subscribe to named topics and receive every message published to them.
class PubSub:
def __init__(self):
self.topics = {}
def subscribe(self, topic, callback):
if topic not in self.topics:
self.topics[topic] = []
self.topics[topic].append(callback)
def publish(self, topic, message):
for callback in self.topics.get(topic, []):
…Redis Pub/Sub Channel Subscribe Mock in Python
A lightweight in-memory mock of Redis pub/sub that lets you subscribe to channels, publish messages, and verify handler behavior in tests without a real Redis server.
class MockRedisPubSub:
def __init__(self):
self.channels = {}
def subscribe(self, channel):
if channel not in self.channels:
self.channels[channel] = []
return self.channels[channel]
def publish(self, channel, message):
if channel in self.channels:
…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 Implement Probabilistic Early Expiration in Python
A Python mock of probabilistic early expiration for caches, using a heap-based expiry queue and random eviction to approximate cache stampede protection.
import heapq
import random
import time
class ProbabilisticEarlyExpirationMock:
def __init__(self, capacity=1024, expiration_probability=0.1):
self.capacity = capacity
self.expiration_probability = expiration_probability
self._items = {}
self._expiry_heap = []
self._next_id…How to Mock Redis Pub/Sub in Python
Test Redis pub/sub logic without a live server using an in-memory fake that queues published messages per channel.
import redis
import time
import threading
class MockRedisPubSub:
def __init__(self):
self.channels = {}
def publish(self, channel, message):
if channel not in self.channels:
return 0
for subscriber in self.channels[channel]:
subscriber.put(message)
ret…How to cache filtered data in Redis with Python
This code caches filtered list results in Redis using an MD5 hash key, returning cached results when available.
import redis
import json
import hashlib
import time
cache = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def filter_data(data, predicate_key, predicate_value):
"""Filter a list of dicts by key-value pair, with Redis caching."""
cache_key = hashlib.md5(
f"{predicate_key}:{pred…Redis LPUSH RPOP List Queue Mock in Python
Implements a FIFO queue using Redis lists with LPUSH and RPOP commands, simulating task processing in Python.
import redis
import time
r = redis.Redis(host='localhost', port=6379, db=0)
queue_key = 'task_queue'
# Push tasks onto the left side (LPUSH)
r.lpush(queue_key, 'task1')
r.lpush(queue_key, 'task2')
r.lpush(queue_key, 'task3')
# Mock processing: pop from the right side (RPOP) — FIFO order
while r.llen(queue_key) > 0:…How to Build a Rate Limiter in Python
Implements a simple sliding-window rate limiter that caps the number of calls per period, used to throttle processing of a data list.
import time
class RateLimiter:
def __init__(self, max_calls, period):
self.max_calls = max_calls
self.period = period
self.timestamps = []
def allow(self):
now = time.time()
self.timestamps = [t for t in self.timestamps if now - t < self.period]
if len(self.tim…How to Implement a Token Bucket Rate Limiter per Client IP in Python
Implements a simple sliding-window rate limiter using a dictionary of timestamp lists per client IP to limit requests per window.
from time import time
from collections import defaultdict
class RateLimiter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.clients = defaultdict(list)
def allow(self, ip: str) -> bool:
now…How to Simulate an Outbox Pattern with Reliable Retry in Python
This code implements a mock outbox pattern with records, delivery attempts, and retries to simulate reliable message publishing.
import time
import itertools
class Outbox:
def __init__(self):
self._records = []
self._seq = itertools.count(1)
def publish(self, topic, payload):
record = {
"id": next(self._seq),
"topic": topic,
"payload": payload,
"status": "pending"…Generate Mock CPU and Memory Metrics in Python
Build a mock_host_metrics() generator that outputs realistic CPU and memory usage percentages for monitoring demos and tests.
import time
import random
def mock_host_metrics():
"""Generate mock CPU and memory metrics for a host."""
cpu_percent = round(random.uniform(10.0, 95.0), 1)
memory_percent = round(random.uniform(20.0, 90.0), 1)
memory_used_mb = round(random.uniform(512, 8192), 1)
return {
"timestamp": in…Generate Synthetic CPU Utilization Metrics in Python
Creates realistic time-series CPU utilization samples with timestamps, noise, and output as structured JSON for observability demos and testing.
from datetime import datetime, timedelta
import random
import json
def generate_metric_samples(base_value, noise, count=60, interval_minutes=1):
"""Generate realistic CPU utilization samples for a given time window."""
timestamps = []
values = []
now = datetime.utcnow()
start_time = now - timede…Generate Synthetic SRE Metrics and Calculate Availability in Python
Create realistic service metrics with random latency, error rate, and request counts, then compute availability and summarize the stream for SLO checks.
from datetime import datetime, timedelta
import random
def generate_service_metrics(service_name: str, minutes: int = 30) -> list[dict]:
"""Generate synthetic SRE metrics for a service across recent minutes."""
metrics = []
now = datetime.now()
for i in range(minutes):
timestamp = now - t…How to Calculate Percentile Latency in Python
Generate mock latency samples with occasional spikes and compute 50th, 90th, 95th, and 99th percentile values in milliseconds.
import random
import statistics
def generate_latency_samples(n=1000):
"""Generate realistic mock latency data (ms) with occasional spikes."""
samples = []
for _ in range(n):
# Normal case: ~50ms with jitter
base = random.gauss(50, 5)
# 2% spike chance: slow downstream or GC pause
…How to Create a TCP DNS Mock Server in Python
This code creates a mock TCP DNS server that listens on a specified port, accepts probe connections, and returns a fixed DNS response header to simulate a live DNS service for testing and observability.
import socket
import threading
def handle_client(client_socket, address):
print(f"[+] Connection from {address}")
try:
while True:
data = client_socket.recv(1024)
if not data:
break
print(f"[*] Received {len(data)} bytes (TCP DNS probe)")
…How to Mock Database Query Duration in Python
Simulate realistic database query durations with random jitter for testing dashboards, alerts, and SLO calculations.
import random
import time
def mock_query_duration(db_name, avg_ms, jitter_ms=5, runs=3):
"""Simulate database query durations with realistic variation."""
durations = []
for _ in range(runs):
# Base duration plus random jitter (can be negative)
duration = avg_ms + random.uniform(-jitter_m…How to Simulate Trace Sampling Head in Python
Simulate head-based probabilistic trace sampling on mock trace data with a configurable sample rate and optional seed for reproducibility.
import random
def trace_sampling_head(mock_traces, sample_rate=0.5, seed=None):
"""Simulate probabilistic trace sampling (head-based) on mock data.
Args:
mock_traces: list of trace dictionaries with a unique 'trace_id'
sample_rate: float 0.0-1.0, probability of keeping a trace
see…Event Sourcing Store in Python: Append-Only Log Mock
Mock an append-only event store in Python — record events, list them, and fetch by ID using a simple list-backed class.
class EventStore:
def __init__(self):
self._events = []
def append(self, event):
event_id = len(self._events) + 1
stored_event = {"id": event_id, "data": event}
self._events.append(stored_event)
return stored_event
def get_events(self):
return list(self._ev…How to Build a Health Check Service Registry in Python
Build a minimal Python service registry that handles registration, deregistration, health checks, and service listing in one simple class.
import random
import time
class ServiceRegistry:
def __init__(self):
self.services = {}
def register(self, name, address):
self.services[name] = {
"address": address,
"status": "healthy",
"registered_at": time.time(),
"checks": 0
}
…How to Build an In-Memory Service Registry Mock in Python
A simple in-memory ServiceRegistry class to register, retrieve, list, and unregister microservice endpoints or configs using a dict, with KeyError guards.
class ServiceRegistry:
def __init__(self):
self._services = {}
def register(self, name, service):
self._services[name] = service
def unregister(self, name):
if name not in self._services:
raise KeyError(f"Service '{name}' not found")
del self._services[name]
…How to Implement an Outbox Pattern Mock in Python
This code demonstrates a simple in-memory outbox pattern mock for publishing domain events and tracking pending events until they are marked as published.
from dataclasses import dataclass, field
from datetime import datetime
from uuid import uuid4
@dataclass
class DomainEvent:
event_id: str = field(default_factory=lambda: str(uuid4()))
occurred_at: datetime = field(default_factory=datetime.utcnow)
class Outbox:
def __init__(self):
self._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.