Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
How to Mock a 202 Accepted Long-Running Operation in Python
Build a mock HTTP server that returns a 202 Accepted response immediately and simulates a long-running operation in the background with threading.
import time
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
class MockHandler(BaseHTTPRequestHandler):
def do_POST(self):
if self.path == "/long-running":
self.send_response(202)
self.send_header("Content-Type", "application/json")
self.end_h…How to Mock a Chunked Encoding Streaming Response in Python
Build a local mock HTTP server with Python's http.server that streams a chunked-encoded response with a 0.5s delay per chunk.
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import time
class ChunkedHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def do_GET(self):
self.send_response(200)
self.send_header("Content-Type", "text/plain")
self.send_header("Transfer-Encoding", "c…How to Validate Request Body JSON Against a Schema in Python
Build a lightweight schema validator to check required fields, types, string lengths, allowed values, and nested objects in a JSON request body.
import json
def validate_against_schema(data, schema, path=""):
errors = []
if not isinstance(data, dict):
errors.append(f"{path}: expected object, got {type(data).__name__}")
return errors
for field, rules in schema.items():
field_path = f"{path}.{field}" if path else field
…Version API by Accept Header with Vendor Media Types in Python
Build a mock HTTP server that routes to API versions by parsing vendor-specific Accept headers in Python.
from http.client import HTTPMessage
from http.server import BaseHTTPRequestHandler, HTTPServer
class VendorVersionHandler(BaseHTTPRequestHandler):
def do_GET(self):
accept = self.headers.get("Accept", "")
version = "v1"
if "application/vnd.myapi.v2+json" in accept:
version = "…How to Build a Flow Control Credit Window in Python
A Python class that reserves, confirms, releases, and settles credit to limit message flow and prevent overload in streaming pipelines.
class CreditWindow:
def __init__(self, max_credit=1000):
self.max_credit = max_credit
self.used_credit = 0
self.pending_credit = 0
def try_reserve(self, amount):
available = self.max_credit - self.used_credit - self.pending_credit
if available >= amount:
…How to Mock NATS Subject Hierarchies with Wildcards in Python
Build a lightweight NATS-style pub/sub mock that matches subject hierarchies with '*' and '>' wildcards for tests or prototypes.
# Mock a simplified NATS subject hierarchy with wildcard matching
# Supports: exact match, '*' (single token), '>' (tail wildcard)
class NATSSubjectMock:
def __init__(self):
self.subscriptions = {} # subject -> list of callbacks
def subscribe(self, subject, callback):
self.subscriptions.setd…How to mock a CQRS projector read model update in Python
Build a CQRS projector class that maintains denormalized read models by applying domain events in a mock order-processing service.
from dataclasses import dataclass, field
from typing import Dict, List, Optional
@dataclass
class OrderReadModel:
order_id: str
customer_name: str
total: float
status: str = "pending"
items: List[Dict] = field(default_factory=list)
def apply_event(self, event_type: str, payload: Dict) -> Non…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, []):
…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 Cache Data in Redis with Python
Build a simple Redis cache wrapper that stores and retrieves JSON data with automatic TTL and serialization.
import redis
import json
import time
class Cache:
def __init__(self, host="localhost", port=6379, db=0, default_ttl=60):
self.client = redis.Redis(host=host, port=port, db=db)
self.default_ttl = default_ttl
def get(self, key):
value = self.client.get(key)
if value is None:
…How to Implement a Redis-Like Cache Dictionary in Python
Build a RedisMockDict class that mimics basic Redis key-value operations with TTL support, expiry cleanup, and standard dict-like methods.
from collections import OrderedDict
import time
class RedisMockDict:
def __init__(self, ttl=None):
self._data = OrderedDict()
self._ttl = ttl # default TTL in seconds, None = no expiry
self._expiry = {}
def set(self, key, value, ttl=None):
"""Set a key-value pair with optiona…How to implement Redlock distributed lock in Python
Simulate Redis Redlock multi-instance locking to show how a distributed lock is acquired only when a majority of instances agree.
import time
import random
import threading
from dataclasses import dataclass
@dataclass
class MockRedisLock:
"""Simple mock of a Redis lock instance."""
name: str
key: str
ttl: int
acquired: bool = False
expires_at: float = 0.0
def acquire(self, sleep_fn=time.sleep):
"""Try to ac…How to implement a write-behind cache with async queue in Python
Build an async write-behind cache that queues writes in memory and flushes them in batches to persistent storage.
import asyncio
from collections import deque
from dataclasses import dataclass
@dataclass
class CacheEntry:
key: str
value: str
class WriteBehindCache:
def __init__(self, flush_interval=1.0):
self.cache = {}
self.queue = deque()
self.flush_interval = flush_interval
self._f…Circuit breaker failure threshold count in Python
Track consecutive or time-windowed failures with a deque to open a circuit breaker and auto-recover to half-open after a cooldown.
from collections import deque
from time import time, sleep
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, recovery_time: float = 10.0):
self.failure_threshold = failure_threshold
self.recovery_time = recovery_time
self.failures: deque[float] = deque()
self.st…How to Cap Retry Attempts in Python with a Decorator
Build a reusable retry decorator that caps attempts, adds delays, and lets flaky services fail fast instead of hanging.
import random
from functools import wraps
from time import sleep
def retry(max_attempts, delay=0.1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
attempts = 0
while attempts < max_attempts:
try:
return func(*args, **kw…How to Implement a Circuit Breaker in Python
A Python dataclass that provides circuit breaker logic with closed, open, and half-open states to fail fast on repeated errors.
from dataclasses import dataclass
from datetime import datetime, timedelta
import time
@dataclass
class CircuitBreaker:
failure_threshold: int = 3
timeout_seconds: float = 5.0
failures: int = 0
state: str = "closed"
last_failure: datetime = None
def call(self, func):
if self.state ==…How to Implement an Adaptive Rate Limiter in Python
Build an adaptive rate limiter that adjusts request intervals dynamically based on recent error rates, slowing down when failures spike.
import time
import random
class AdaptiveRateLimiter:
"""Simple adaptive rate limiter that reduces requests when error rate is high."""
def __init__(self, min_interval=0.1, max_interval=2.0, error_threshold=0.3):
self.min_interval = min_interval
self.max_interval = max_interval
sel…How to Mock a Circuit Breaker Reset Timeout in Python
This code implements a simple circuit breaker with a reset timeout test, simulating a flaky service to show half-open state transitions.
import time
import random
class CircuitBreaker:
def __init__(self, failure_threshold=3, reset_timeout=5):
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED" # CLOSED (nor…Implement a Circuit Breaker Pattern in Python
This code implements a simple circuit breaker that opens after a threshold of consecutive failures, causing subsequent calls to fail fast without invoking the underlying function.
class CircuitBreaker:
def __init__(self, failure_threshold=3):
self.failure_threshold = failure_threshold
self.failure_count = 0
self.open = False
def call(self, func, *args, **kwargs):
if self.open:
raise RuntimeError("Circuit is open - failing fast")
try:
…How to Build a Burn Rate Alert with Multiple Time Windows in Python
Track token consumption and trigger alerts when the burn rate exceeds a threshold across multiple time windows using deque and time-based sliding windows.
import time
from collections import deque
class BurnRateAlert:
def __init__(self, windows_seconds=(60, 300, 900), threshold_rate=0.8):
self.windows = {w: deque() for w in windows_seconds}
self.threshold_rate = threshold_rate
self.previous_tokens = None
def record_sample(self, current_…How to Build a Python Latency Histogram with Mock Buckets
This code implements a mock latency histogram that records request durations into configurable buckets and outputs counts, total, and average latency.
import time
import random
from collections import Counter
class LatencyHistogram:
def __init__(self, buckets):
self.buckets = sorted(buckets)
self.counts = Counter()
self.total = 0
self.sum_latency = 0
def record(self, latency_ms):
for i, boundary in enumerate(self.bu…How to Build an HTTP Server Request Duration Histogram in Python
Create a small HTTP server that times each GET request, buckets the duration, and prints a histogram on shutdown.
import time
import random
from collections import Counter
from http.server import HTTPServer, BaseHTTPRequestHandler
class HistogramHandler(BaseHTTPRequestHandler):
response_times = Counter()
def do_GET(self):
start = time.perf_counter()
time.sleep(random.uniform(0.001, 0.1))
duratio…Summary Quantile Mock Sketch in Python
Build a memory-efficient sketch that stores sorted bins of data points to answer approximate quantile queries like median without keeping all values in memory.
import random
import statistics
from collections import Counter
class SummaryQuantileSketch:
"""
A simple sketch that stores a fixed-size summary of data (min, max, deciles)
using sorted bins, then answers approximate quantile queries.
"""
def __init__(self, bins=10):
self.bins = bins
…Backward Compatible Schema Evolution in Python
A mock schema validator that evolves JSON schemas while preserving backward compatibility by keeping old fields and validating required ones.
import json
from copy import deepcopy
class SchemaValidator:
def __init__(self, schema):
self.schema = schema
def evolve(self, new_schema):
"""Evolve mock schema while keeping backward compatibility."""
for field in self.schema:
if field not in new_schema:
…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.