Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Mock an API Key Header Authentication Server in Python
A minimal HTTP server that validates requests using an X-API-Key header and returns JSON responses for authenticated and unauthenticated calls.
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
API_KEYS = {"test-user": "secret-key-123"}
class AuthHandler(BaseHTTPRequestHandler):
def do_GET(self):
auth = self.headers.get("X-API-Key")
if not auth or auth not in API_KEYS.values():
self.send_response…How to Poll an Operation Status Endpoint in Python
Mock a polling endpoint in Python that simulates checking an async operation's status until it completes or times out.
import time
import random
def poll_status(url: str, timeout: float = 5.0) -> dict:
"""Mock a polling endpoint that eventually returns a completed status."""
start = time.time()
while time.time() - start < timeout:
# Simulate delayed response
time.sleep(0.2)
# 80% chance to report …How to handle CORS preflight OPTIONS requests in Python
Create a mock HTTP server with a CORS preflight OPTIONS handler that returns the correct headers for browser-based API requests.
from http.server import BaseHTTPRequestHandler, HTTPServer
class CORSRequestHandler(BaseHTTPRequestHandler):
def _send_cors_headers(self):
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
self.send_head…How to mock a REST POST endpoint in Python
Create a simple mock REST server that responds to POST requests with a 201 status and a JSON body.
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
class MockHandler(BaseHTTPRequestHandler):
def do_POST(self):
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length) if content_length else b"{}"
try:
data = json…Implement If-Match Precondition Update in Python
A mock resource store that uses the If-Match header's ETag to guard updates, preventing overwrites from stale clients.
from dataclasses import dataclass
from typing import Optional
@dataclass
class Resource:
id: str
version: int = 1
data: str = ""
etag: str = "etag-1"
class MockResourceStore:
def __init__(self):
self.resources = {}
def update(self, resource_id: str, new_data: str, if_match: Optiona…Serve Swagger UI with Python's built-in HTTP server
Hosts a self-contained Swagger UI with a mock OpenAPI spec using only Python's standard library HTTP server.
from http.server import HTTPServer, SimpleHTTPRequestHandler
import os
import tempfile
SWAGGER_HTML = """<!DOCTYPE html>
<html>
<head>
<title>Mock Swagger UI</title>
<link rel="stylesheet" href="https://unpkg.com/swagger-ui-dist@4/swagger-ui.css">
</head>
<body>
<div id="swagger-ui"></div>
<script src…How to Implement Publish-Subscribe Fanout with Multiple Subscribers in Python
Create a simple publish-subscribe system in Python that broadcasts messages to multiple subscriber callbacks for a given topic.
import time
class PubSub:
def __init__(self):
self.subscribers = {}
def subscribe(self, topic, callback):
if topic not in self.subscribers:
self.subscribers[topic] = []
self.subscribers[topic].append(callback)
def publish(self, topic, message):
if topic in sel…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 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 Offset Commit Auto vs Manual in Python
Demonstrates a Kafka-style offset commit function with auto/manual modes and tests it using unittest.mock.patch.
from unittest.mock import Mock, patch
def commit_offsets(topic_partition_offsets, auto_commit=False):
"""Manually commit offsets or simulate auto-commit."""
if auto_commit:
print(f"Auto-committing offsets: {topic_partition_offsets}")
return {"status": "auto_committed"}
print(f"Manuall…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:
…Coalescing duplicate in-flight requests: one shared result for concurrent callers
Runs identical concurrent requests through a single shared call, caching the result while it's in flight and returning the same value to all callers.
import time
import threading
from collections import defaultdict
class CoalescingExecutor:
def __init__(self):
self._locks = defaultdict(threading.Lock)
self._in_flight = {}
def execute(self, key, func):
with self._locks[key]:
if key in self._in_flight:
re…How to Mock Redis EXPIRE, TTL, and PERSIST in Python
A lightweight in-memory MockRedis class that simulates Redis key expiration, TTL, and persist behavior for tests and local development.
import time
class MockRedis:
def __init__(self):
self._store = {}
self._expiry = {}
def set(self, key, value):
self._store[key] = value
self._expiry.pop(key, None)
return True
def expire(self, key, ttl_seconds):
if key not in self._store:
retur…How to Mock zlib Compression for Cache Values in Python
Compress cache values with zlib and mock the compress function in unit tests to simulate cache behavior.
import zlib
from unittest.mock import patch
def compress_value(data: bytes) -> bytes:
"""Compress data using zlib and return the compressed bytes."""
return zlib.compress(data)
def decompress_value(compressed: bytes) -> bytes:
"""Decompress zlib data and return the original bytes."""
return zlib.deco…How to create a stable cache key from function arguments in Python
Generate a stable SHA-256 cache key from normalized function arguments, with keyword order normalized and tests using mocks.
import hashlib
import json
from unittest.mock import Mock
def make_cache_key(*args, **kwargs):
"""Normalize args/kwargs into a stable hash key for caching."""
normalized = {
"args": [repr(arg) for arg in args],
"kwargs": {key: repr(value) for key, value in sorted(kwargs.items())}
}
pa…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:…Build a queue-based admission control system in Python
Implement a simple bounded-queue admission controller that accepts or rejects incoming requests based on current queue capacity.
from collections import deque
import time
class AdmissionControl:
"""Simple admission control using a bounded queue.
Requests arrive at the queue; they are admitted in FIFO order.
If the queue is full, the incoming request is rejected.
"""
def __init__(self, capacity: int):
self.capacit…Fixed Window Counter Rate Limiting in Python
A simple fixed window counter rate limiter that allows a maximum number of requests per 60-second window, with a mock time simulation.
from collections import deque
from time import time
class FixedWindowCounter:
def __init__(self, max_requests):
self.max_requests = max_requests
self.window_start = int(time())
self.window_count = 0
def allow_request(self):
current_time = int(time())
if current_time >=…How to Implement Hedged Requests in Python
This code demonstrates a hedged request pattern using threading, which sends duplicate calls and returns the first result that arrives within a timeout.
import time
from unittest.mock import Mock
def hedged_request(call, timeout=0.05):
"""Execute two duplicate calls, return first result within timeout."""
result_container = {}
def run_and_store():
result_container['result'] = call()
result_container['done'] = True
# Simulate slow cal…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 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 Stop Receiving Requests Until Ready in Python
A mock server that refuses requests until a readiness gate is passed, simulating fail-stop behavior for production reliability.
import random
import time
class MockServer:
def __init__(self):
self.ready = False
self.requests_received = 0
def readiness_check(self):
"""Simulates a readiness probe. Returns True only when ready."""
if not self.ready:
return False
return True
def r…How to implement rate limiting per API key in Python
A simple sliding-window rate limiter that tracks request timestamps per API key and rejects requests exceeding the configured limit.
import time
API_RATE_LIMITS = {"api_key_1": 5, "api_key_2": 3} # max requests per window
WINDOW_SECONDS = 10
class RateLimiter:
def __init__(self, limits, window):
self.limits = limits
self.window = window
self.requests = {key: [] for key in limits}
def allow(self, api_key):
…How to mock a fallback return value in Python
Test a function that returns a default value on failure by mocking requests.get and its side effects.
from unittest.mock import Mock, patch
import requests
def fetch_data(url, default=None):
try:
response = requests.get(url)
response.raise_for_status()
return response.json()
except (requests.RequestException, ValueError):
return default
with patch("requests.get") as mock_get:
…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.