Microservices patterns
Service boundaries, discovery, inter-service calls, and decomposition patterns.
How to Mock a Service Registry in Python with an In-Memory Dict
A lightweight ServiceRegistry class backed by a dict, exposing register, unregister, lookup, list, and health-check methods.
class ServiceRegistry:
def __init__(self):
self._services = {}
def register(self, name, endpoint, version="1.0"):
self._services[name] = {
"endpoint": endpoint,
"version": version,
"status": "healthy"
}
def unregister(self, name):
return…How to Mock an API Gateway Router in Python
Create a lightweight HTTP server that routes requests to mock microservice responses, simulating an API gateway for local development and testing.
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
class SimpleGateway(BaseHTTPRequestHandler):
def do_GET(self):
routes = {
"/users": {"service": "user-service", "status": "ok", "count": 42},
"/orders": {"service": "order-service", "status": "ok", "count": 17}…How to Mock an Ambassador Edge Proxy in Python
Build a lightweight mock Ambassador edge proxy with Python's http.server that responds to health and user endpoint requests for local development and testing.
import http.server
import json
import urllib.parse
import threading
class AmbassadorProxyHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
parsed = urllib.parse.urlparse(self.path)
if parsed.path == "/health":
self.send_response(200)
self.send_header("Content-T…How to Order Partition Key Events in Python (Mock Stream)
Generate a mock event stream grouped by partition key and sort it deterministically by key then sequence in Python.
import itertools
import random
def partition_key_events(keys, events_per_key=3, seed=None):
"""Produce a realistic-looking, but mock, event stream grouped by partition key.
Args:
keys: iterable of partition keys (e.g. strings or ints).
events_per_key: how many events we want per key.
…How to implement a circuit breaker in Python
A Python CircuitBreaker class that tracks failures, opens after a threshold, and retries after a timeout.
class CircuitBreaker:
def __init__(self, failure_threshold=3, timeout=5):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED"
def call(self, mock_downstream):
if self.state …How to implement read-your-writes sticky routing in Python
A mock StickyRouter class that routes all requests for the same key to the same node, ensuring read-after-write consistency.
import random
class StickyRouter:
def __init__(self, nodes):
self.nodes = nodes
self.routes = {}
def route(self, key):
if key not in self.routes:
self.routes[key] = random.choice(self.nodes)
return self.routes[key]
def read(self, key):
node = self.rout…How to implement round-robin load balancing in Python
Implement a client-side round-robin load balancer that distributes requests sequentially across a list of mock servers using itertools.cycle.
import itertools
import random
class MockServer:
def __init__(self, name):
self.name = name
def handle_request(self, request_id):
return f"Server {self.name} handled request #{request_id}"
class RoundRobinLoadBalancer:
def __init__(self, servers):
self.servers = servers
…Idempotent Consumer Event Processing in Python
Track processed event IDs to skip duplicates and count event types for a reliable, idempotent consumer.
import json
from collections import defaultdict
class EventProcessor:
def __init__(self):
self.processed_ids = set()
self.counts = defaultdict(int)
def process_event(self, event):
event_id = event["id"]
if event_id in self.processed_ids:
return {"status": "skipped"…JWT Service-to-Service Authentication Mock in Python
Create and verify HS256 JWTs for service-to-service authentication without external libraries.
import hashlib
import hmac
import base64
import json
import time
class JWTMock:
"""Minimal JWT service-to-service mock using HS256."""
def __init__(self, secret):
self.secret = secret.encode()
@staticmethod
def _b64url_encode(data):
return base64.urlsafe_b64encode(data).rstr…Mock a Sidecar Logger with Python Metrics
Simulate a sidecar logger that tracks request counts, error rates, and endpoint hits, producing a metrics snapshot.
import random
import time
from collections import defaultdict
class SidecarLogger:
def __init__(self):
self.metrics = defaultdict(int)
self.total_requests = 0
self.error_count = 0
def log_request(self, endpoint, status_code):
"""Simulate logging a request and updating metrics…Python Saga Compensating Steps Mock
Mock a distributed transaction saga with forward steps and compensating actions that reverse partial progress on failure.
from datetime import datetime
def make_payment(user_id, amount):
print(f"[{datetime.now():%H:%M:%S}] Payment of ${amount} processed for user {user_id}")
return {"step": "payment", "status": "ok", "details": f"${amount} charged"}
def deduct_inventory(order_id, items):
print(f"[{datetime.now():%H:%M:%S}]…Retry idempotent GET requests in Python
A Python function that retries an idempotent GET request a fixed number of times with a delay between attempts, raising a RuntimeError only after all retries fail.
import time
import urllib.error
import urllib.request
from http.client import HTTPException
def fetch_with_retry(url, max_retries=3, delay=1.0):
for attempt in range(1, max_retries + 1):
try:
with urllib.request.urlopen(url, timeout=5) as response:
return response.read().decode…Saga pattern orchestration with rollback in Python
Orchestrate a distributed transaction with Saga steps and automated compensation rollback on failure.
import time
import random
class SagaStep:
def __init__(self, name):
self.name = name
self.executed = False
def execute(self):
print(f"Executing {self.name}...")
time.sleep(0.2)
if random.random() < 0.3:
raise RuntimeError(f"{self.name} failed")
sel…Scatter Gather Aggregate Pattern in Python
Simulates a scatter/gather/aggregate pattern by distributing work across items, gathering results, and aggregating them.
import random
def process_items(items, scatter_fn, gather_fn, aggregate_fn):
"""Simple scatter/gather/aggregate pattern simulation."""
scattered = [scatter_fn(item) for item in items]
gathered = [gather_fn(item) for item in scattered]
return aggregate_fn(gathered)
if __name__ == "__main__":
data …Strangler Fig Migration Pattern in Python
Gradually reroute calls from a legacy service to a modern replacement using a runtime switch and feature detection.
from dataclasses import dataclass
@dataclass
class PaymentService:
def process(self, amount: float) -> str:
return f"Legacy processed ${amount:.2f}"
class StranglerFig:
def __init__(self):
self._new_service = None
def attach_new(self, service):
self._new_service = service
de…Browse by section
Each section groups closely related Python snippets.
Microservices patterns — Python code examples
What you will find here
This page collects microservices patterns snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.
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.