Microservices Architecture: Principles, Patterns, and Best Practices

# Example: E-commerce microservices decomposition

# User Service - User management and authentication
class UserService:
    def create_user(self, email: str, name: str) -> User:
        pass
    
    def authenticate(self, email: str, password: str) -> Token:
        pass

# Product Service - Product catalog
class ProductService:
    def get_product(self, product_id: str) -> Product:
        pass
    
    def search_products(self, query: str) -> List[Product]:
        pass

# Order Service - Order management
class OrderService:
    def create_order(self, user_id: str, items: List[OrderItem]) -> Order:
        pass
    
    def get_order(self, order_id: str) -> Order:
        pass

# Inventory Service - Stock management
class InventoryService:
    def reserve_stock(self, items: List[StockItem]) -> bool:
        pass
    
    def release_stock(self, reservation_id: str):
        pass

# Payment Service - Payment processing
class PaymentService:
    def process_payment(self, order_id: str, amount: float) -> Payment:
        pass

# Order bounded context owns order data
class Order:
    def __init__(self, order_id: str, user_id: str, items: List[OrderItem]):
        self.order_id = order_id
        self.user_id = user_id
        self.items = items
        self.status = "pending"
    
    def calculate_total(self) -> float:
        return sum(item.total for item in self.items)

# Synchronous: REST call
class OrderService:
    def __init__(self, http_client, product_service_url: str):
        self.http_client = http_client
        self.product_url = product_service_url
    
    def create_order(self, user_id: str, items: List[dict]):
        # Check product availability synchronously
        for item in items:
            response = self.http_client.get(
                f"{self.product_url}/products/{item['product_id']}"
            )
            if response.json()["stock"] < item["quantity"]:
                raise OutOfStockError(item["product_id"])
        # Create order...

# Asynchronous: Event-driven
class OrderService:
    def __init__(self, event_bus):
        self.event_bus = event_bus
    
    def create_order(self, user_id: str, items: List[dict]):
        order = self._create_order(user_id, items)
        
        # Publish event for async processing
        self.event_bus.publish(
            "order.created",
            {"order_id": order.id, "items": items}
        )
        return order

class APIGateway:
    """Single entry point for all client requests."""
    
    def __init__(self):
        self.routes = {
            "/api/users": "http://user-service:8080",
            "/api/products": "http://product-service:8080",
            "/api/orders": "http://order-service:8080",
        }
    
    async def route_request(self, method: str, path: str, body: dict):
        for prefix, service_url in self.routes.items():
            if path.startswith(prefix):
                return await self._forward(method, service_url + path, body)
        return {"error": "Not found"}, 404