microservices-patterns

name	microservices-patterns
description	Design microservices architectures with service boundaries, event-driven communication, and resilience patterns. Use when building distributed systems, decomposing monoliths, or implementing microservices. Use when this capability is needed.
metadata	{"author":"baotoq"}

Microservices Patterns

Master microservices architecture patterns including service boundaries, inter-service communication, data management, and resilience patterns for building distributed systems.

When to Use This Skill

• Decomposing monoliths into microservices
• Designing service boundaries and contracts
• Implementing inter-service communication
• Managing distributed data and transactions
• Building resilient distributed systems
• Implementing service discovery and load balancing
• Designing event-driven architectures

Core Concepts

1. Service Decomposition Strategies

By Business Capability

• Organize services around business functions
• Each service owns its domain
• Example: OrderService, PaymentService, InventoryService

By Subdomain (DDD)

• Core domain, supporting subdomains
• Bounded contexts map to services
• Clear ownership and responsibility

Strangler Fig Pattern

• Gradually extract from monolith
• New functionality as microservices
• Proxy routes to old/new systems

2. Communication Patterns

Synchronous (Request/Response)

• REST APIs
• gRPC
• GraphQL

Asynchronous (Events/Messages)

• Event streaming (Kafka)
• Message queues (RabbitMQ, SQS)
• Pub/Sub patterns

3. Data Management

Database Per Service

• Each service owns its data
• No shared databases
• Loose coupling

Saga Pattern

• Distributed transactions
• Compensating actions
• Eventual consistency

4. Resilience Patterns

Circuit Breaker

• Fail fast on repeated errors
• Prevent cascade failures

Retry with Backoff

• Transient fault handling
• Exponential backoff

Bulkhead

• Isolate resources
• Limit impact of failures

Service Decomposition Patterns

Pattern 1: By Business Capability

# E-commerce example

# Order Service
class OrderService:
    """Handles order lifecycle."""

    async def create_order(self, order_data: dict) -> Order:
        order = Order.create(order_data)

        # Publish event for other services
        await self.event_bus.publish(
            OrderCreatedEvent(
                order_id=order.id,
                customer_id=order.customer_id,
                items=order.items,
                total=order.total
            )
        )

        return order

# Payment Service (separate service)
class PaymentService:
    """Handles payment processing."""

    async def process_payment(self, payment_request: PaymentRequest) -> PaymentResult:
        # Process payment
        result = await self.payment_gateway.charge(
            amount=payment_request.amount,
            customer=payment_request.customer_id
        )

        if result.success:
            await self.event_bus.publish(
                PaymentCompletedEvent(
                    order_id=payment_request.order_id,
                    transaction_id=result.transaction_id
                )
            )

        return result

# Inventory Service (separate service)
class InventoryService:
    """Handles inventory management."""

    async def reserve_items(self, order_id: , items: [OrderItem]) -> ReservationResult:
        
         item  items:
            available =  .inventory_repo.get_available(item.product_id)
             available < item.quantity:
                 ReservationResult(
                    success=,
                    error=
                )

        
        reservation =  .create_reservation(order_id, items)

         .event_bus.publish(
            InventoryReservedEvent(
                order_id=order_id,
                reservation_id=reservation.
            )
        )

         ReservationResult(success=, reservation=reservation)

Pattern 2: API Gateway

from fastapi import FastAPI, HTTPException, Depends
import httpx
from circuitbreaker import circuit

app = FastAPI()

class APIGateway:
    """Central entry point for all client requests."""

    def __init__(self):
        self.order_service_url = "http://order-service:8000"
        self.payment_service_url = "http://payment-service:8001"
        self.inventory_service_url = "http://inventory-service:8002"
        self.http_client = httpx.AsyncClient(timeout=5.0)

    @circuit(failure_threshold=5, recovery_timeout=30)
    async def call_order_service(self, path: str, method: str = "GET", **kwargs):
        """Call order service with circuit breaker."""
        response = await self.http_client.request(
            method,
            f"{self.order_service_url}{path}",
            **kwargs
        )
        response.raise_for_status()
        return response.json()

    async def create_order_aggregate(self, order_id: str) -> dict:
        """Aggregate data from multiple services."""
        
        order, payment, inventory =  asyncio.gather(
            .call_order_service(),
            .call_payment_service(),
            .call_inventory_service(),
            return_exceptions=
        )

        
        result = {: order}
          (payment, Exception):
            result[] = payment
          (inventory, Exception):
            result[] = inventory

         result


  ():
    
    :
        
        order =  gateway.call_order_service(
            ,
            method=,
            json=order_data
        )
         {: order}
     httpx.HTTPError  e:
         HTTPException(status_code=, detail=)

Communication Patterns

Pattern 1: Synchronous REST Communication

# Service A calls Service B
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class ServiceClient:
    """HTTP client with retries and timeout."""

    def __init__(self, base_url: str):
        self.base_url = base_url
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(5.0, connect=2.0),
            limits=httpx.Limits(max_keepalive_connections=20)
        )

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    async def get(self, path: str, **kwargs):
        """GET with automatic retries."""
        response = await self.client.get(f"{self.base_url}{path}", **kwargs)
        response.raise_for_status()
        return response.json()

    async def post(self, path: str, **kwargs):
        """POST request."""
        response = await self.client.post(f"", **kwargs)
        response.raise_for_status()
         response.json()


payment_client = ServiceClient()
result =  payment_client.post(, json=payment_data)

Pattern 2: Asynchronous Event-Driven

# Event-driven communication with Kafka
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
import json
from dataclasses import dataclass, asdict
from datetime import datetime

@dataclass
class DomainEvent:
    event_id: str
    event_type: str
    aggregate_id: str
    occurred_at: datetime
    data: dict

class EventBus:
    """Event publishing and subscription."""

    def __init__(self, bootstrap_servers: List[str]):
        self.bootstrap_servers = bootstrap_servers
        self.producer = None

    async def start(self):
        self.producer = AIOKafkaProducer(
            bootstrap_servers=self.bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode()
        )
        await self.producer.start()

    async def publish(self, event: DomainEvent):
        """Publish event to Kafka topic."""
        topic = event.event_type
        await self.producer.send_and_wait(
            topic,
            value=asdict(event),
            key=event.aggregate_id.encode()
        )

    async  ():
        
        consumer = AIOKafkaConsumer(
            topic,
            bootstrap_servers=.bootstrap_servers,
            value_deserializer= v: json.loads(v.decode()),
            group_id=
        )
         consumer.start()

        :
              message  consumer:
                event_data = message.value
                 handler(event_data)
        :
             consumer.stop()


  ():
    order =  save_order(order_data)

    event = DomainEvent(
        event_id=(uuid.uuid4()),
        event_type=,
        aggregate_id=order.,
        occurred_at=datetime.now(),
        data={
            : order.,
            : order.customer_id,
            : order.total
        }
    )

     event_bus.publish(event)


  ():
    
    order_id = event_data[][]
    items = event_data[][]

    
     reserve_inventory(order_id, items)

Pattern 3: Saga Pattern (Distributed Transactions)

# Saga orchestration for order fulfillment
from enum import Enum
from typing import List, Callable

class SagaStep:
    """Single step in saga."""

    def __init__(
        self,
        name: str,
        action: Callable,
        compensation: Callable
    ):
        self.name = name
        self.action = action
        self.compensation = compensation

class SagaStatus(Enum):
    PENDING = "pending"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    FAILED = "failed"

class OrderFulfillmentSaga:
    """Orchestrated saga for order fulfillment."""

    def __init__(self):
        self.steps: List[SagaStep] = [
            SagaStep(
                "create_order",
                action=self.create_order,
                compensation=self.cancel_order
            ),
            SagaStep(
                "reserve_inventory",
                action=self.reserve_inventory,
                compensation=self.release_inventory
            ),
            SagaStep(
                "process_payment",
                action=self.process_payment,
                compensation=.refund_payment
            ),
            SagaStep(
                ,
                action=.confirm_order,
                compensation=.cancel_order_confirmation
            )
        ]

      () -> SagaResult:
        
        completed_steps = []
        context = {: order_data}

        :
             step  .steps:
                
                result =  step.action(context)
                  result.success:
                    
                     .compensate(completed_steps, context)
                     SagaResult(
                        status=SagaStatus.FAILED,
                        error=result.error
                    )

                completed_steps.append(step)
                context.update(result.data)

             SagaResult(status=SagaStatus.COMPLETED, data=context)

         Exception  e:
            
             .compensate(completed_steps, context)
             SagaResult(status=SagaStatus.FAILED, error=(e))

      ():
        
         step  (completed_steps):
            :
                 step.compensation(context)
             Exception  e:
                
                ()

    
      () -> StepResult:
        order =  order_service.create(context[])
         StepResult(success=, data={: order.})

      ():
         order_service.cancel(context[])

      () -> StepResult:
        result =  inventory_service.reserve(
            context[],
            context[][]
        )
         StepResult(
            success=result.success,
            data={: result.reservation_id}
        )

      ():
         inventory_service.release(context[])

      () -> StepResult:
        result =  payment_service.charge(
            context[],
            context[][]
        )
         StepResult(
            success=result.success,
            data={: result.transaction_id},
            error=result.error
        )

      ():
         payment_service.refund(context[])

Resilience Patterns

Circuit Breaker Pattern

from enum import Enum
from datetime import datetime, timedelta
from typing import Callable, Any

class CircuitState(Enum):
    CLOSED = "closed"  # Normal operation
    OPEN = "open"      # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

class CircuitBreaker:
    """Circuit breaker for service calls."""

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 30,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold

        self.failure_count = 0
        self.success_count = 0
        self.state = CircuitState.CLOSED
        self.opened_at = None

    async def call(self, func: Callable, *args, **kwargs) -> Any:
        """Execute function with circuit breaker."""

         .state == CircuitState.OPEN:
             ._should_attempt_reset():
                .state = CircuitState.HALF_OPEN
            :
                 CircuitBreakerOpenError()

        :
            result =  func(*args, **kwargs)
            ._on_success()
             result

         Exception  e:
            ._on_failure()
            

     ():
        
        .failure_count = 

         .state == CircuitState.HALF_OPEN:
            .success_count += 
             .success_count >= .success_threshold:
                .state = CircuitState.CLOSED
                .success_count = 

     ():
        
        .failure_count += 

         .failure_count >= .failure_threshold:
            .state = CircuitState.OPEN
            .opened_at = datetime.now()

         .state == CircuitState.HALF_OPEN:
            .state = CircuitState.OPEN
            .opened_at = datetime.now()

     () -> :
        
         (
            datetime.now() - .opened_at
            > timedelta(seconds=.recovery_timeout)
        )


breaker = CircuitBreaker(failure_threshold=, recovery_timeout=)

  ():
      breaker.call(
        payment_client.process_payment,
        payment_data
    )

Resources

• references/service-decomposition-guide.md: Breaking down monoliths
• references/communication-patterns.md: Sync vs async patterns
• references/saga-implementation.md: Distributed transactions
• assets/circuit-breaker.py: Production circuit breaker
• assets/event-bus-template.py: Kafka event bus implementation
• assets/api-gateway-template.py: Complete API gateway

Best Practices

1. Service Boundaries: Align with business capabilities
2. Database Per Service: No shared databases
3. API Contracts: Versioned, backward compatible
4. Async When Possible: Events over direct calls
5. Circuit Breakers: Fail fast on service failures
6. Distributed Tracing: Track requests across services
7. Service Registry: Dynamic service discovery
8. Health Checks: Liveness and readiness probes

Common Pitfalls

• Distributed Monolith: Tightly coupled services
• Chatty Services: Too many inter-service calls
• Shared Databases: Tight coupling through data
• No Circuit Breakers: Cascade failures
• Synchronous Everything: Tight coupling, poor resilience
• Premature Microservices: Starting with microservices
• Ignoring Network Failures: Assuming reliable network
• No Compensation Logic: Can't undo failed transactions

---

Converted and distributed by TomeVault — claim your Tome and manage your conversions.