Microservices patterns
Service boundaries, discovery, inter-service calls, and decomposition patterns.
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…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 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 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…How to Mock Service Call Timeouts in Python
Simulate service calls with configurable timeouts using Mock to patch sleep and randomness, covering success and timeout cases.
import time
from unittest.mock import Mock, patch
# Simulate a service call with configurable timeout
def call_service(service_name, timeout=5):
"""Mock a service call that may time out."""
start = time.time()
print(f"Calling {service_name}...")
# Simulate service latency (randomized for realism)…How to Mock a Choreography Saga in Python
Simulate a choreography-based saga with event envelopes, status tracking, and compensating actions to model distributed transactions.
import json
from dataclasses import dataclass, asdict
from typing import List, Optional
from enum import Enum
class SagaStatus(Enum):
PENDING = "PENDING"
COMPLETING = "COMPLETING"
COMPLETED = "COMPLETED"
FAILED = "FAILED"
@dataclass
class EventEnvelope:
event_type: str
order_id: str
sta…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…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}]…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.