Microservices Architecture Development: Scalable Backend System Design

This microservices architecture implementation shows service discovery, API gateway, and event-driven communication patterns used in production systems.

import asyncio
import json
from typing import Dict, List, Optional
from dataclasses import dataclass
from abc import ABC, abstractmethod
import aiohttp
import redis
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ServiceConfig:
    name: str
    host: str
    port: int
    version: str
    health_check_path: str = "/health"
    dependencies: List[str] = None

class ServiceRegistry:
    """Service registry for microservices discovery"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.services: Dict[str, ServiceConfig] = {}
    
    async def register_service(self, service: ServiceConfig):
        """Register a microservice"""
        service_key = f"service:{service.name}:{service.version}"
        service_data = {
            "host": service.host,
            "port": service.port,
            "version": service.version,
            "health_check_path": service.health_check_path,
            "dependencies": service.dependencies or []
        }
        
        await self.redis.hset(service_key, mapping=service_data)
        await self.redis.expire(service_key, 300)  # 5-minute TTL
        
        logger.info(f"Registered service: {service.name} v{service.version}")
    
    async def discover_service(self, service_name: str, version: str = None) -> Optional[ServiceConfig]:
        """Discover a microservice"""
        if version:
            service_key = f"service:{service_name}:{version}"
        else:
            # Find latest version
            pattern = f"service:{service_name}:*"
            keys = await self.redis.keys(pattern)
            if not keys:
                return None
            
            # Get the latest version
            latest_key = max(keys, key=lambda k: k.decode().split(':')[-1])
            service_key = latest_key.decode()
        
        service_data = await self.redis.hgetall(service_key)
        if not service_data:
            return None
        
        return ServiceConfig(
            name=service_name,
            host=service_data[b'host'].decode(),
            port=int(service_data[b'port']),
            version=service_data[b'version'].decode(),
            health_check_path=service_data[b'health_check_path'].decode(),
            dependencies=json.loads(service_data[b'dependencies'].decode())
        )

class APIGateway:
    """API Gateway for microservices routing"""
    
    def __init__(self, service_registry: ServiceRegistry):
        self.registry = service_registry
        self.routes: Dict[str, str] = {}
        self.circuit_breakers: Dict[str, CircuitBreaker] = {}
    
    def add_route(self, path: str, service_name: str, version: str = None):
        """Add a route to a microservice"""
        self.routes[path] = f"{service_name}:{version or 'latest'}"
        self.circuit_breakers[service_name] = CircuitBreaker()
    
    async def route_request(self, path: str, method: str, headers: Dict, body: str = None) -> Dict:
        """Route request to appropriate microservice"""
        if path not in self.routes:
            raise HTTPException(status_code=404, detail="Route not found")
        
        service_info = self.routes[path]
        service_name, version = service_info.split(':')
        
        # Discover service
        service = await self.registry.discover_service(service_name, version)
        if not service:
            raise HTTPException(status_code=503, detail="Service unavailable")
        
        # Check circuit breaker
        circuit_breaker = self.circuit_breakers[service_name]
        if circuit_breaker.is_open():
            raise HTTPException(status_code=503, detail="Service circuit breaker open")
        
        try:
            # Make request to microservice
            url = f"http://{service.host}:{service.port}{path}"
            
            async with aiohttp.ClientSession() as session:
                async with session.request(
                    method=method,
                    url=url,
                    headers=headers,
                    data=body
                ) as response:
                    result = await response.json()
                    circuit_breaker.record_success()
                    return result
        
        except Exception as e:
            circuit_breaker.record_failure()
            logger.error(f"Request to {service_name} failed: {str(e)}")
            raise HTTPException(status_code=503, detail="Service unavailable")

class CircuitBreaker:
    """Circuit breaker pattern for fault tolerance"""
    
    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    def is_open(self) -> bool:
        """Check if circuit breaker is open"""
        if self.state == "OPEN":
            if asyncio.get_event_loop().time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
                return False
            return True
        return False
    
    def record_success(self):
        """Record successful request"""
        self.failure_count = 0
        self.state = "CLOSED"
    
    def record_failure(self):
        """Record failed request"""
        self.failure_count += 1
        self.last_failure_time = asyncio.get_event_loop().time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"

class EventBus:
    """Event bus for microservices communication"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.subscribers: Dict[str, List[callable]] = {}
    
    async def publish_event(self, event_type: str, event_data: Dict):
        """Publish an event"""
        event = {
            "type": event_type,
            "data": event_data,
            "timestamp": asyncio.get_event_loop().time()
        }
        
        await self.redis.publish(f"events:{event_type}", json.dumps(event))
        logger.info(f"Published event: {event_type}")
    
    async def subscribe_to_event(self, event_type: str, handler: callable):
        """Subscribe to an event"""
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        
        self.subscribers[event_type].append(handler)
        
        # Start listening for events
        pubsub = self.redis.pubsub()
        await pubsub.subscribe(f"events:{event_type}")
        
        async for message in pubsub.listen():
            if message['type'] == 'message':
                event = json.loads(message['data'])
                await handler(event)
    
    async def handle_event(self, event: Dict):
        """Handle incoming event"""
        event_type = event['type']
        if event_type in self.subscribers:
            for handler in self.subscribers[event_type]:
                try:
                    await handler(event)
                except Exception as e:
                    logger.error(f"Event handler failed: {str(e)}")

class MicroserviceBase(ABC):
    """Base class for microservices"""
    
    def __init__(self, name: str, version: str, host: str, port: int):
        self.name = name
        self.version = version
        self.host = host
        self.port = port
        self.app = FastAPI(title=f"{name} Service", version=version)
        self.redis = redis.Redis(host='localhost', port=6379, db=0)
        self.registry = ServiceRegistry(self.redis)
        self.event_bus = EventBus(self.redis)
        
        # Setup health check endpoint
        self.app.get("/health")(self.health_check)
    
    @abstractmethod
    async def startup(self):
        """Service startup logic"""
        pass
    
    @abstractmethod
    async def shutdown(self):
        """Service shutdown logic"""
        pass
    
    async def health_check(self):
        """Health check endpoint"""
        return {"status": "healthy", "service": self.name, "version": self.version}
    
    async def register_service(self, dependencies: List[str] = None):
        """Register this service"""
        service_config = ServiceConfig(
            name=self.name,
            host=self.host,
            port=self.port,
            version=self.version,
            dependencies=dependencies
        )
        
        await self.registry.register_service(service_config)
    
    async def start_service(self):
        """Start the microservice"""
        await self.startup()
        await self.register_service()
        
        logger.info(f"Started {self.name} service on {self.host}:{self.port}")
        
        # Start FastAPI server
        import uvicorn
        uvicorn.run(self.app, host=self.host, port=self.port)

# Example microservice implementations
class UserService(MicroserviceBase):
    def __init__(self):
        super().__init__("user-service", "1.0.0", "0.0.0.0", 8001)
        self.users: Dict[str, Dict] = {}
        self.setup_routes()
    
    def setup_routes(self):
        """Setup API routes"""
        @self.app.post("/users")
        async def create_user(user_data: dict):
            user_id = f"user_{len(self.users) + 1}"
            self.users[user_id] = user_data
            return {"id": user_id, **user_data}
        
        @self.app.get("/users/{user_id}")
        async def get_user(user_id: str):
            if user_id not in self.users:
                raise HTTPException(status_code=404, detail="User not found")
            return self.users[user_id]
    
    async def startup(self):
        """Startup logic"""
        await self.event_bus.subscribe_to_event("user.created", self.handle_user_created)
    
    async def shutdown(self):
        """Shutdown logic"""
        pass
    
    async def handle_user_created(self, event: Dict):
        """Handle user created event"""
        logger.info(f"User created event received: {event['data']}")

class OrderService(MicroserviceBase):
    def __init__(self):
        super().__init__("order-service", "1.0.0", "0.0.0.0", 8002)
        self.orders: Dict[str, Dict] = {}
        self.setup_routes()
    
    def setup_routes(self):
        """Setup API routes"""
        @self.app.post("/orders")
        async def create_order(order_data: dict):
            order_id = f"order_{len(self.orders) + 1}"
            self.orders[order_id] = order_data
            
            # Publish order created event
            await self.event_bus.publish_event("order.created", {
                "order_id": order_id,
                "user_id": order_data.get("user_id"),
                "amount": order_data.get("amount")
            })
            
            return {"id": order_id, **order_data}
    
    async def startup(self):
        """Startup logic"""
        pass
    
    async def shutdown(self):
        """Shutdown logic"""
        pass

# Example usage
async def main():
    """Example microservices setup"""
    
    # Start user service
    user_service = UserService()
    user_task = asyncio.create_task(user_service.start_service())
    
    # Start order service
    order_service = OrderService()
    order_task = asyncio.create_task(order_service.start_service())
    
    # Wait for services to start
    await asyncio.gather(user_task, order_task)

if __name__ == "__main__":
    asyncio.run(main())