Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
Circuit Breaker Pattern in Python for LLM API Calls
Implements a circuit breaker class that wraps LLM client calls to fail fast when the service is degrading, then recover automatically after a timeout.
import time
class CircuitBreaker:
def __init__(self, failure_threshold=3, recovery_timeout=5):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.state = "closed"
self.last_failure_time = None
def call(self, …Build a Complete Website Sitemap Generator Without External Services
Crawl a website recursively using only Python's standard library to generate a structured sitemap of internal links.
import json
from urllib.parse import urlparse, urljoin
from collections import deque
import urllib.request
import urllib.error
import re
from html.parser import HTMLParser
class SitemapParser(HTMLParser):
def __init__(self, base_url):
super().__init__()
self.base_url = base_url
self.links …How to Mock Azure Key Vault Secret Get in Python
Mock an Azure Key Vault client's get_secret method with unittest.mock to test functions that retrieve secret values without hitting the real service.
import unittest
from unittest.mock import MagicMock, patch
def get_secret(key_vault_client, secret_name):
"""Retrieve a secret value from an Azure Key Vault client."""
secret = key_vault_client.get_secret(secret_name)
return secret.value
class TestKeyVaultSecretGet(unittest.TestCase):
def test_get_…How to Mock RDS Snapshot Create and Restore in Python
Mock AWS RDS snapshot creation and restore operations in Python tests using moto and boto3 without hitting real AWS services.
import boto3
from moto import mock_rds
@mock_rds
def create_and_restore_snapshot():
client = boto3.client("rds", region_name="us-east-1")
client.create_db_instance(
DBInstanceIdentifier="my-db",
DBInstanceClass="db.t3.micro",
Engine="postgres",
AllocatedStorage=20,
Mas…How to Run an Integration Test with Docker Compose Mock in Python
Run a Python integration test against a docker-compose environment, using mocks to simulate service health and business logic responses.
import subprocess
import json
from typing import Dict
def run_integration_test() -> Dict[str, str]:
"""
Simulates an integration test against a docker-compose environment
using a mock service that returns canned responses.
"""
# Mock docker-compose environment check
env_ready = subprocess.run(…Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States
Implement a circuit breaker with closed, open, and half-open states to prevent repeated calls to failing services and allow recovery after a timeout.
class CircuitBreaker:
def __init__(self, failure_threshold=3, timeout_seconds=5):
self.failure_threshold = failure_threshold
self.timeout_seconds = timeout_seconds
self.state = "closed"
self.failure_count = 0
self.last_failure_time = None
def record_success(self):
…How to implement saga orchestration with compensating steps in Python
Orchestrate a distributed transaction across services, rolling back completed steps with compensations when a later step fails.
class InventoryService:
def reserve(self, order_id):
print(f"[Inventory] Reserving stock for order {order_id}")
return True
def compensate(self, order_id):
print(f"[Inventory] Releasing stock for order {order_id}")
class PaymentService:
def charge(self, order_id):
print(f…Implement Bulkhead Thread Pool Isolation in Python
Create isolated thread pools with a bulkhead pattern to protect different services from cascading failures.
import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor
class Bulkhead:
"""Simple bulkhead isolation: separate thread pools for different tasks."""
def __init__(self, max_workers):
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self.active = …Lazy loading with a proxy in Python: defer expensive service creation
A lazy proxy defers creating an expensive service object until its method is first called, then caches it for reuse.
import time
import random
class ExpensiveService:
def __init__(self, name):
self.name = name
print(f"Creating expensive service: {self.name}")
def fetch_data(self):
time.sleep(1)
return f"Data from {self.name}: {random.randint(1, 100)}"
class LazyProxy:
def __init__(sel…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…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 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…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…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)")
…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,…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…Distributed tracing with contextvars in Python
Propagate trace and span IDs across function calls using contextvars to mock distributed tracing in a single process.
import contextvars
import uuid
import time
_trace_context = contextvars.ContextVar("trace_context", default=None)
class TraceContext:
def __init__(self, trace_id, parent_span_id):
self.trace_id = trace_id
self.parent_span_id = parent_span_id
self.span_id = uuid.uuid4().hex[:16]
s…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…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.