Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
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…Health Check Mark Unhealthy Stop Traffic Mock in Python
Simulates a health check with a 20% failure rate and automatically stops traffic when the service is unhealthy.
import time
import random
class HealthCheck:
def __init__(self):
self.is_healthy = True
self.stop_traffic = False
def check_health(self):
# Simulate health check with random failure rate (20% chance unhealthy)
self.is_healthy = random.random() > 0.2
return self.is_heal…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 Inject Random Latency for Chaos Testing in Python
Mock unreliable services by wrapping functions with a decorator that adds random network-like delays before execution.
import random
import time
from functools import wraps
def inject_latency(func):
@wraps(func)
def wrapper(*args, **kwargs):
latency = random.uniform(0.1, 0.5)
print(f"Injecting {latency:.3f}s latency...")
time.sleep(latency)
return func(*args, **kwargs)
return wrapper
@inje…How to Mock Fault Injection Percentage in Python
Simulate a service with a 30% failure rate using random.random to test error handling and retries.
import random
class Service:
def call(self):
if random.random() < 0.3: # 30% failure rate
raise ConnectionError("Simulated network fault")
return "ok"
def main():
svc = Service()
random.seed(42) # deterministic for demonstration
results = []
for _ in range(10):
…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…How to Mock a Slow Startup Probe in Python
Simulate slow service initialization with a configurable mock delay to test readiness probes.
import time
from dataclasses import dataclass, field
@dataclass
class StartupProbe:
name: str
min_wait_sec: float = 0.5
max_wait_sec: float = 2.0
_ready: bool = field(default=False, init=False, repr=False)
def initialize(self) -> None:
"""Simulate slow startup with a fixed mock delay."""…Saga Compensating Transaction Mock in Python
Simulates a distributed transaction using a saga pattern with compensating actions that roll back steps on failure.
import random
import time
class OrderService:
def __init__(self):
self.orders = {}
def create_order(self, order_id):
print(f"[Order] Creating order {order_id}...")
time.sleep(0.1)
if random.random() < 0.3: # 30% chance of failure
raise RuntimeError(f"Order {order…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 Check Service Readiness Dependencies in Python
This code simulates a readiness check for external dependencies (database, cache, queue) with mock availability data and reports readiness status.
import sys
from datetime import datetime
def check_dependencies(config):
results = []
for dep, required in config.items():
available = mock_availability(dep)
status = "READY" if available >= required else "NOT READY"
results.append((dep, available, required, status))
return result…How to Create a Deployment Environment Tag in Python
Generate a standardized deployment tag string by combining service and environment names with an f-string.
def mock_env_tag(service, environment):
return f"{service}-{environment}"
if __name__ == "__main__":
service = "api-gateway"
environment = "production"
tag = mock_env_tag(service, environment)
print(f"Deployment tag: {tag}")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 Generate and Propagate W3C Trace Context Headers in Python
Generate and propagate W3C traceparent and tracestate headers for distributed tracing in Python, with mock service headers.
import uuid
def generate_w3c_traceparent(trace_id=None, parent_id=None, flags="01"):
if trace_id is None:
trace_id = uuid.uuid4().hex[:32]
if parent_id is None:
parent_id = uuid.uuid4().hex[:16]
return f"00-{trace_id}-{parent_id}-{flags}"
def create_mock_headers(service_name, trace_id=N…How to Mock HTTP Client Latency in Python
Simulate outbound HTTP request latency with configurable ranges to test timeouts, retries, and SLO monitoring without external services.
import time
import random
def mock_latency(host: str, min_ms: int = 100, max_ms: int = 500) -> dict:
"""Simulate an outbound HTTP request with mock latency."""
latency_ms = random.randint(min_ms, max_ms)
start = time.perf_counter()
time.sleep(latency_ms / 1000)
elapsed_ms = (time.perf_counter() - …How to Mock Service Resource Attributes in Python
Temporarily override service name, version, and other resource attributes with a context manager, then restore them automatically.
from contextlib import contextmanager
import random
_SERVICE_ATTRIBUTES = {
"service.name": "payment-api",
"service.version": "1.4.2",
"service.instance.id": str(random.randint(10000, 99999)),
"service.namespace": "production",
}
@contextmanager
def mock_service_attributes(**overrides):
"""Tempor…How to Parse Log Lines with Regex in Python
Extracts timestamp, log level, service name, and message from a log line using compiled regex named groups.
import re
LOG_PATTERN = re.compile(
r'^(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) '
r'\[(?P<level>\w+)\] '
r'\((?P<service>[^)]+)\) '
r'(?P<message>.*)$'
)
def parse_log_line(line: str) -> dict:
match = LOG_PATTERN.match(line)
if not match:
return {"error": "invalid log format…Mock Health Endpoint Liveness Check in Python
Simulate a liveness endpoint that reports service health with a configurable failure rate and uptime.
import time
import random
def liveness_check(service_name: str, failure_rate: float = 0.1) -> dict:
"""Mock health check that returns liveness status with a configurable failure rate."""
healthy = random.random() > failure_rate
response = {
"service": service_name,
"status": "alive" if he…BFF aggregation pattern: combine multiple service responses in Python
Mock three backend services and aggregate their responses into one unified payload — the BFF pattern every Python microservice gateway relies on.
from dataclasses import dataclass
from typing import Any
@dataclass
class Service:
name: str
data: dict[str, Any]
def get_user_service() -> Service:
return Service("user", {"id": 1, "name": "Alice"})
def get_orders_service() -> Service:
return Service("orders", {"total": 299.99, "count": 2})
de…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:
…Bulkhead Thread Pool per Service Mock in Python
Simulates a bulkhead pattern with per-service thread pools and semaphore-based rejection to isolate failures between dependent services.
import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor
class ServiceBulkhead:
def __init__(self, name, max_threads, max_queue):
self.name = name
self.executor = ThreadPoolExecutor(max_workers=max_threads)
self.semaphore = threading.Semaphore(max_thread…CQRS with Separate Read and Write Repositories in Python
Implement CQRS in Python with separate write and read repositories, using commands for mutations and frozen DTOs for queries.
from dataclasses import dataclass
from typing import Dict, List, Optional
# --- Write side: commands mutate state ---
@dataclass
class CreateUserCommand:
id: int
name: str
class UserWriteRepository:
def __init__(self) -> None:
self._store: Dict[int, Dict[str, object]] = {}
def create(self,…Cache-Aside Pattern in Python: Per-Service Mock
A Python mock of the cache-aside pattern for a single microservice—lazy-load from a database into an in-memory cache and invalidate on updates.
class ServiceCache:
def __init__(self):
self.database = {"user:1": "Alice", "user:2": "Bob", "user:3": "Charlie"}
self.cache = {}
def get_user(self, user_id):
cache_key = f"user:{user_id}"
if cache_key in self.cache:
print(f"CACHE HIT: {cache_key}")
retu…Consumer Driven Contract Pact Mock in Python
Define and verify consumer-driven contracts using Pact's Consumer and Provider classes, mocking the provider to assert expected interactions.
from pact import Consumer, Provider
pact = Consumer('OrderService').has_pact_with(Provider('InventoryService'))
@Pact.verify()
class TestInventoryContract:
def test_get_inventory(self):
expected = {"item": "widget", "quantity": 100}
(pact
.given('inventory exists for widget')
.u…Correlation ID HTTP header mock in Python
A lightweight HTTP server that echoes or generates correlation IDs to help test distributed systems.
import json
import uuid
from http.server import BaseHTTPRequestHandler, HTTPServer
class CorrelationHandler(BaseHTTPRequestHandler):
CORRELATION_HEADER = "X-Correlation-ID"
def do_GET(self):
correlation_id = self.headers.get(self.CORRELATION_HEADER) or str(uuid.uuid4())
response = {
…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.