Moving from monolith to microservices was one of my most challenging architectural transitions. While microservices offer scalability and team autonomy, they introduce significant complexity. This article covers the patterns I've successfully used in production microservices architectures.
Service Decomposition
When to Use Microservices
Good candidates:
Large teams (>50 engineers) working on same codebase
Different components needing independent scaling
Services with distinct deployment cadences
Teams wanting technology diversity
Stay with monolith if:
Small team (<10 engineers)
Early-stage product with rapidly changing requirements
Simple domain that fits in one codebase
Team lacks distributed systems expertise
Domain-Driven Design
Inter-Service Communication
Synchronous (HTTP/gRPC)
Asynchronous (Message Queue)
Circuit Breaker Pattern
Service Discovery
API Gateway Pattern
Saga Pattern (Distributed Transactions)
Lessons Learned
What worked:
Start with monolith, extract services as needed
Event-driven architecture for loose coupling
Circuit breakers for resilience
API gateway for centralized concerns
Comprehensive monitoring and tracing
What didn't work:
Micro-services from day one (premature optimization)
# Bounded contexts - each microservice owns its domain
# User Service
class UserService:
"""Owns user authentication and profile data."""
def create_user(self, email, password):
pass
def authenticate(self, email, password):
pass
# Order Service
class OrderService:
"""Owns order processing and fulfillment."""
def create_order(self, user_id, items):
pass
def get_order_status(self, order_id):
pass
# Payment Service
class PaymentService:
"""Owns payment processing."""
def process_payment(self, order_id, amount):
pass
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
class OrderServiceClient:
"""Client for calling Order Service."""
def __init__(self, base_url: str):
self.base_url = base_url
self.client = httpx.AsyncClient(timeout=10.0)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10)
)
async def get_order(self, order_id: str) -> dict:
"""Get order with automatic retries."""
response = await self.client.get(f"{self.base_url}/orders/{order_id}")
response.raise_for_status()
return response.json()
# Order Service publishes events
class OrderService:
def __init__(self, event_bus):
self.event_bus = event_bus
async def create_order(self, user_id: str, items: list):
order = self._create_order_in_db(user_id, items)
# Publish event
await self.event_bus.publish('order.created', {
'order_id': order['id'],
'user_id': user_id,
'total': order['total']
})
return order
# Payment Service subscribes to events
class PaymentService:
def __init__(self, event_bus):
self.event_bus = event_bus
self.event_bus.subscribe('order.created', self.handle_order_created)
async def handle_order_created(self, event: dict):
"""Process payment when order is created."""
order_id = event['order_id']
total = event['total']
await self.process_payment(order_id, total)
from enum import Enum
from datetime import datetime, timedelta
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing if service recovered
class CircuitBreaker:
"""
Circuit breaker to prevent cascading failures.
I use this for all external service calls.
"""
def __init__(
self,
failure_threshold: int = 5,
timeout_duration: int = 60,
success_threshold: int = 2
):
self.failure_threshold = failure_threshold
self.timeout_duration = timeout_duration
self.success_threshold = success_threshold
self.state = CircuitState.CLOSED
self.failures = 0
self.successes = 0
self.last_failure_time = None
async def call(self, func, *args, **kwargs):
"""Execute function with circuit breaker protection."""
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
else:
raise Exception("Circuit breaker is OPEN")
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
"""Handle successful call."""
self.failures = 0
if self.state == CircuitState.HALF_OPEN:
self.successes += 1
if self.successes >= self.success_threshold:
self.state = CircuitState.CLOSED
self.successes = 0
def _on_failure(self):
"""Handle failed call."""
self.failures += 1
self.last_failure_time = datetime.utcnow()
if self.failures >= self.failure_threshold:
self.state = CircuitState.OPEN
def _should_attempt_reset(self) -> bool:
"""Check if enough time has passed to try again."""
return (
self.last_failure_time and
datetime.utcnow() - self.last_failure_time > timedelta(seconds=self.timeout_duration)
)
# Usage
circuit_breaker = CircuitBreaker()
async def call_external_service():
return await circuit_breaker.call(external_api_call)
# Kubernetes service discovery (DNS-based)
# Services automatically discover each other via DNS
# service.yaml
"""
apiVersion: v1
kind: Service
metadata:
name: order-service
spec:
selector:
app: order-service
ports:
- port: 8080
targetPort: 8080
"""
# Other services can reach it at:
# http://order-service.default.svc.cluster.local:8080
# Python client
class ServiceDiscovery:
"""Service discovery for Kubernetes."""
def get_service_url(self, service_name: str, namespace: str = "default") -> str:
"""Get service URL from Kubernetes DNS."""
return f"http://{service_name}.{namespace}.svc.cluster.local:8080"
# Consul-based service discovery
import consul
class ConsulServiceDiscovery:
"""Service discovery using Consul."""
def __init__(self, consul_host: str = "localhost"):
self.consul = consul.Consul(host=consul_host)
def register_service(self, name: str, port: int, health_check_url: str):
"""Register service with Consul."""
self.consul.agent.service.register(
name=name,
service_id=f"{name}-{port}",
port=port,
check=consul.Check.http(health_check_url, interval="10s")
)
def discover_service(self, name: str) -> str:
"""Discover service endpoint."""
_, services = self.consul.health.service(name, passing=True)
if services:
service = services[0]
address = service['Service']['Address']
port = service['Service']['Port']
return f"http://{address}:{port}"
raise Exception(f"Service {name} not found")
from fastapi import FastAPI, HTTPException
import httpx
app = FastAPI(title="API Gateway")
class APIGateway:
"""
API Gateway that routes to microservices.
Handles auth, rate limiting, and request routing.
"""
def __init__(self):
self.services = {
"users": "http://user-service:8080",
"orders": "http://order-service:8080",
"payments": "http://payment-service:8080"
}
self.client = httpx.AsyncClient()
async def route_request(
self,
service: str,
path: str,
method: str = "GET",
**kwargs
):
"""Route request to appropriate microservice."""
if service not in self.services:
raise HTTPException(status_code=404, detail="Service not found")
url = f"{self.services[service]}/{path}"
response = await self.client.request(method, url, **kwargs)
return response.json()
gateway = APIGateway()
@app.get("/api/users/{user_id}")
async def get_user(user_id: str):
"""Route to user service."""
return await gateway.route_request("users", f"users/{user_id}")
@app.get("/api/orders/{order_id}")
async def get_order(order_id: str):
"""Route to order service."""
return await gateway.route_request("orders", f"orders/{order_id}")
class OrderSaga:
"""
Saga pattern for coordinating distributed transaction.
"""
def __init__(self, services):
self.user_service = services['user']
self.inventory_service = services['inventory']
self.payment_service = services['payment']
self.order_service = services['order']
async def execute(self, order_data: dict):
"""Execute saga with compensating transactions."""
compensations = []
try:
# Step 1: Validate user
user = await self.user_service.get_user(order_data['user_id'])
if not user:
raise Exception("User not found")
# Step 2: Reserve inventory
await self.inventory_service.reserve(order_data['items'])
compensations.append(
lambda: self.inventory_service.release(order_data['items'])
)
# Step 3: Process payment
payment = await self.payment_service.charge(
order_data['user_id'],
order_data['total']
)
compensations.append(
lambda: self.payment_service.refund(payment['id'])
)
# Step 4: Create order
order = await self.order_service.create(order_data)
return order
except Exception as e:
# Execute compensating transactions in reverse order
for compensation in reversed(compensations):
try:
await compensation()
except Exception as comp_error:
# Log compensation failure
print(f"Compensation failed: {comp_error}")
raise
