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 mTLS Between Services in Python
Simulate mutual TLS authentication between two services using Python's ssl module with self-signed certificates.
import ssl
import socket
import threading
import tempfile
from pathlib import Path
import subprocess
def create_test_cert(cert_path: Path, key_path: Path, common_name: str = "localhost"):
"""Generate a self-signed certificate using openssl."""
subprocess.run([
"openssl", "req", "-x509", "-newkey", "rs…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 the Database per service pattern in Python
Simulate separate databases per microservice in Python using dataclasses and in-memory dictionaries, showing how services own their data independently.
import json
from dataclasses import dataclass, asdict
from typing import Dict, List
@dataclass
class User:
id: int
name: str
email: str
@dataclass
class Order:
id: int
user_id: int
product: str
amount: float
class UserServiceDB:
"""Simulates a separate database for the User servic…How to mock an external service in Python with an anti-corruption facade
This code implements an anti-corruption facade that mocks an external API, allowing client code to interact with a simulated service while keeping the same interface.
class AntiCorruptionFacade:
"""Mocks a real API while keeping the same interface."""
def __init__(self, data_store):
self._data_store = data_store
self._calls = []
def get_user(self, user_id):
self._calls.append(f"get_user({user_id})")
return self._data_store.get(u…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…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…Zero Trust Service Auth Mock in Python
A simple HMAC-based token issuance and validation mock that enforces zero trust between microservices.
import hmac
import hashlib
import json
import time
class ZeroTrustAuth:
def __init__(self, secret_key):
self.secret_key = secret_key
self.service_tokens = {}
def issue_token(self, service_name, ttl=300):
payload = {
"service": service_name,
"issued_at": int(tim…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.