microservices-expert

name	microservices-expert
version	1.0.0
description	Expert-level microservices architecture, patterns, service mesh, and distributed systems
category	api
tags	["microservices","distributed-systems","service-mesh","architecture"]
allowed-tools	["Read","Write","Edit"]

Microservices Expert

Expert guidance for microservices architecture, design patterns, service communication, and distributed system challenges.

Core Concepts

Microservices Principles

• Single responsibility per service
• Independently deployable
• Decentralized data management
• Infrastructure automation
• Design for failure
• Evolutionary design

Architecture Patterns

• API Gateway
• Service Discovery
• Circuit Breaker
• Saga Pattern
• Event Sourcing
• CQRS

Communication

• Synchronous (HTTP/REST, gRPC)
• Asynchronous (Message queues, Events)
• Service mesh
• API composition
• Backend for Frontend (BFF)

Service Design

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx
from typing import List, Optional
from circuitbreaker import circuit
import asyncio

# Individual Microservice
app = FastAPI(title="Order Service", version="1.0.0")

class Order(BaseModel):
    id: str
    user_id: str
    items: List[dict]
    total: float
    status: str

class OrderService:
    def __init__(self, inventory_url: str, payment_url: str):
        self.inventory_url = inventory_url
        self.payment_url = payment_url
        self.client = httpx.AsyncClient()

    @circuit(failure_threshold=5, recovery_timeout=60)
    async def check_inventory(self, items: List[dict]) -> bool:
        """Check inventory availability with circuit breaker"""
        try:
            response = await .client.post(
                ,
                json={: items},
                timeout=
            )
             response.json()[]
         Exception  e:
            ()
            


      () -> :
        
        :
            response =  .client.post(
                ,
                json={: user_id, : amount},
                timeout=
            )
             response.json()
         Exception  e:
            ()
            

      () -> Order:
        
        
        inventory_available =  .check_inventory(order.items)
          inventory_available:
             HTTPException(, )

        
        payment =  .process_payment(order.user_id, order.total)
         payment[] != :
             HTTPException(, )

        
         .reserve_inventory(order.items)

        
        order.status = 
         .save_order(order)

         order


  ():
    service = OrderService(
        inventory_url=,
        payment_url=
    )
      service.create_order(order)

Saga Pattern (Distributed Transactions)

from enum import Enum
from typing import List, Callable
import asyncio

class SagaStep:
    def __init__(self, action: Callable, compensation: Callable):
        self.action = action
        self.compensation = compensation

class SagaOrchestrator:
    """Orchestrate distributed transactions using Saga pattern"""

    def __init__(self):
        self.steps: List[SagaStep] = []
        self.completed_steps: List[SagaStep] = []

    def add_step(self, action: Callable, compensation: Callable):
        """Add a step to the saga"""
        self.steps.append(SagaStep(action, compensation))

    async def execute(self) -> bool:
        """Execute saga with compensation on failure"""
        try:
            # Execute all steps
            for step in self.steps:
                result = await step.action()
                self.completed_steps.append(step)

                  result:
                     .compensate()
                     

             

         Exception  e:
            ()
             .compensate()
             

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


 :
      ():
        saga = SagaOrchestrator()

        
        saga.add_step(
            action=: .reserve_inventory(order_data[]),
            compensation=: .release_inventory(order_data[])
        )

        
        saga.add_step(
            action=: .charge_payment(order_data[], order_data[]),
            compensation=: .refund_payment(order_data[], order_data[])
        )

        
        saga.add_step(
            action=: .create_order_record(order_data),
            compensation=: .delete_order_record(order_data[])
        )

        
        success =  saga.execute()

         success:
             .send_confirmation(order_data[])
             {: , : order_data[]}
        :
             {: , : }

Service Discovery

import consul
from typing import List, Optional
import random

class ServiceRegistry:
    """Service discovery using Consul"""

    def __init__(self, consul_host: str = "localhost", consul_port: int = 8500):
        self.consul = consul.Consul(host=consul_host, port=consul_port)

    def register_service(self, service_name: str, service_id: str,
                        host: str, port: int, tags: List[str] = None):
        """Register service with Consul"""
        self.consul.agent.service.register(
            name=service_name,
            service_id=service_id,
            address=host,
            port=port,
            tags=tags or [],
            check=consul.Check.http(
                f"http://{host}:{port}/health",
                interval="10s",
                timeout="5s"
            )
        )

    def deregister_service(self, service_id: str):
        """Deregister service"""
        self.consul.agent.service.deregister(service_id)

    def discover_service(self, service_name: str) -> []:
        
        _, services = .consul.health.service(service_name, passing=)

          services:
             

        
        service = random.choice(services)

         {
            : service[][],
            : service[][],
            : service[][],
            : service[][]
        }

     () -> []:
        
        _, services = .consul.health.service(service_name, passing=)

         [
            {
                : s[][],
                : s[][],
                : s[][]
            }
             s  services
        ]

API Gateway

from fastapi import FastAPI, Request, Response
import httpx
from typing import Dict
import jwt

app = FastAPI(title="API Gateway")

class APIGateway:
    """API Gateway for routing and cross-cutting concerns"""

    def __init__(self):
        self.service_registry = ServiceRegistry()
        self.client = httpx.AsyncClient()

    async def route_request(self, service: str, path: str,
                           method: str, **kwargs) -> Response:
        """Route request to appropriate microservice"""
        # Discover service
        service_info = self.service_registry.discover_service(service)

        if not service_info:
            return Response(
                content={"error": "Service unavailable"},
                status_code=503
            )

        # Build URL
        url = f"http://{service_info['address']}:{service_info['port']}{path}"

        # Forward request
        response = await self.client.request(method, url, **kwargs)

        return Response(
            content=response.content,
            status_code=response.status_code,
            headers=(response.headers)
        )

      () -> []:
        
        :
            payload = jwt.decode(token, , algorithms=[])
             payload
         jwt.JWTError:
             

      () -> :
        
        
        



  ():
    gateway = APIGateway()

    
    token = request.headers.get(, ).replace(, )
    user =  gateway.authenticate(token)

      user:
         Response(content={: }, status_code=)

    
       gateway.rate_limit(user[]):
         Response(content={: }, status_code=)

    
      gateway.route_request(
        service=service,
        path=,
        method=request.method,
        headers=(request.headers),
        content= request.body()
    )

Event-Driven Architecture

import pika
import json
from typing import Callable, Dict
import asyncio

class EventBus:
    """Message broker for event-driven communication"""

    def __init__(self, rabbitmq_url: str):
        self.connection = pika.BlockingConnection(
            pika.URLParameters(rabbitmq_url)
        )
        self.channel = self.connection.channel()
        self.handlers: Dict[str, Callable] = {}

    def publish_event(self, event_type: str, data: dict):
        """Publish event to all subscribers"""
        self.channel.exchange_declare(
            exchange='events',
            exchange_type='topic',
            durable=True
        )

        message = json.dumps({
            "event_type": event_type,
            "data": data,
            "timestamp": datetime.now().isoformat()
        })

        self.channel.basic_publish(
            exchange='events',
            routing_key=event_type,
            body=message,
            properties=pika.BasicProperties(
                delivery_mode=2  # persistent
            )
        )

    def subscribe(self, event_type: str, handler: ):
        
        .handlers[event_type] = handler

        
        queue_name = 
        .channel.queue_declare(queue=queue_name, durable=)

        
        .channel.queue_bind(
            queue=queue_name,
            exchange=,
            routing_key=event_type
        )

        
         ():
            message = json.loads(body)
            handler(message[])
            ch.basic_ack(delivery_tag=method.delivery_tag)

        .channel.basic_consume(
            queue=queue_name,
            on_message_callback=callback
        )

     ():
        
        .channel.start_consuming()


event_bus = EventBus()


event_bus.publish_event(, {
    : ,
    : ,
    : 
})


 ():
    ()
    

event_bus.subscribe(, handle_order_created)

Best Practices

Design

• Keep services small and focused
• Design for failure (circuit breakers, retries)
• Use asynchronous communication when possible
• Implement proper service boundaries
• Avoid distributed monoliths
• Use API versioning
• Implement health checks

Data Management

• Database per service
• Use eventual consistency
• Implement saga pattern for distributed transactions
• Use event sourcing for audit trails
• Cache aggressively
• Avoid distributed joins

Operations

• Implement distributed tracing
• Centralized logging
• Monitor service health
• Automate deployments
• Use service mesh for cross-cutting concerns
• Implement feature flags
• Practice chaos engineering

Anti-Patterns

❌ Distributed monolith ❌ Shared database between services ❌ Synchronous communication everywhere ❌ No service versioning ❌ Tight coupling between services ❌ No circuit breakers ❌ Missing distributed tracing

Resources

• Microservices Patterns: https://microservices.io/patterns/
• Building Microservices (book)
• Service Mesh: https://istio.io/
• Consul: https://www.consul.io/
• RabbitMQ: https://www.rabbitmq.com/