API Gateway Communication — Centralized Entry Point for Microservices

# gateway_responsibilities.py
# API gateway responsibilities

def gateway_responsibilities():
    print("API Gateway Core Responsibilities")
    print("=" * 40)
    print()
    
    responsibilities = [
        {
            "function": "Request Routing",
            "desc": "Route incoming requests to the appropriate microservice based on URL path, headers, or method.",
            "example": "/api/users/* -> user-service, /api/orders/* -> order-service"
        },
        {
            "function": "Authentication",
            "desc": "Validate tokens and authenticate requests before they reach backend services.",
            "example": "Validate JWT, extract user ID, attach to forwarded headers"
        },
        {
            "function": "Rate Limiting",
            "desc": "Limit requests from a client to prevent abuse and protect backend services.",
            "example": "100 requests/minute per API key, 429 response when exceeded"
        },
        {
            "function": "Response Aggregation",
            "desc": "Combine responses from multiple services into a single response to reduce client round trips.",
            "example": "GET /order-detail/123 -> gateway calls order + payment + shipping services"
        },
        {
            "function": "Protocol Translation",
            "desc": "Convert between external and internal protocols (REST <-> gRPC, HTTP/1.1 <-> HTTP/2).",
            "example": "External REST client -> gateway translates to internal gRPC calls"
        },
    ]
    
    for r in responsibilities:
        print(f"{r['function']:30s}")
        print(f"  {r['desc']}")
        print(f"  Example: {r['example']}")
        print()

gateway_responsibilities()

# routing.py
# API gateway request routing

def request_routing():
    print("API Gateway Request Routing")
    print("=" * 40)
    print()
    
    routing_code = """
class ApiGateway:
    def __init__(self, registry):
        self.registry = registry
        self.routes = [
            Route("/api/users/", "user-service", ["GET", "POST", "PUT", "DELETE"]),
            Route("/api/orders/", "order-service", ["GET", "POST"]),
            Route("/api/orders/*/cancel", "order-service", ["POST"]),
            Route("/api/payments/", "payment-service", ["GET", "POST"]),
            Route("/api/files/", "file-service", ["GET", "POST", "DELETE"]),
        ]
    
    def route_request(self, request):
        # Find matching route
        for route in self.routes:
            if self._matches(request, route):
                return self._forward(request, route)
        
        return {"error": "route not found"}, 404
    
    def _matches(self, request, route):
        # Check path prefix and HTTP method
        path_matches = request.path.startswith(route.path_pattern) or \
                       self._wildcard_match(request.path, route.path_pattern)
        method_matches = request.method in route.methods
        return path_matches and method_matches
    
    def _forward(self, request, route):
        service_name = route.service
        instances = self.registry.get_instances(service_name)
        
        if not instances:
            return {"error": f"{service_name} unavailable"}, 503
        
        # Select instance and forward
        instance = self._select_instance(instances)
        target_url = f"http://{instance['host']}:{instance['port']}{request.path}"
        
        # Add gateway metadata to headers
        headers = dict(request.headers)
        headers["X-Forwarded-By"] = "api-gateway"
        headers["X-User-Id"] = request.user_id  # Set by auth middleware
        
        return self._proxy(target_url, request.method, headers, request.body)
    
    def _wildcard_match(self, path, pattern):
        parts = path.split("/")
        pat_parts = pattern.split("/")
        if len(parts) != len(pat_parts):
            return False
        for p, pp in zip(parts, pat_parts):
            if pp != "*" and p != pp:
                return False
        return True
"""
    print(routing_code)

request_routing()

# aggregation.py
# Response aggregation pattern

def response_aggregation():
    print("Response Aggregation")
    print("=" * 40)
    print()
    
    code = """
import asyncio
import httpx

class AggregationGateway:
    """
    Aggregates responses from multiple services 
    to reduce client round trips.
    """
    
    async def get_order_detail(self, order_id, user_id):
        # Call multiple services in parallel
        async with httpx.AsyncClient() as client:
            order_task = self._get_order(client, order_id)
            payment_task = self._get_payment(client, order_id)
            shipping_task = self._get_shipping(client, order_id)
            
            order, payment, shipping = await asyncio.gather(
                order_task, payment_task, shipping_task,
                return_exceptions=True
            )
        
        # Build aggregated response
        response = {
            "order": order if not isinstance(order, Exception) else None,
            "payment": payment if not isinstance(payment, Exception) else None,
            "shipping": shipping if not isinstance(shipping, Exception) else None,
        }
        
        # Add error info for any failed services
        errors = {}
        if isinstance(order, Exception):
            errors["order"] = str(order)
        if isinstance(payment, Exception):
            errors["payment"] = str(payment)
        if isinstance(shipping, Exception):
            errors["shipping"] = str(shipping)
        
        if errors:
            response["_errors"] = errors
        
        return response
    
    async def _get_order(self, client, order_id):
        resp = await client.get(
            f"http://order-service/api/orders/{order_id}",
            timeout=5.0
        )
        resp.raise_for_status()
        return resp.json()
    
    async def _get_payment(self, client, order_id):
        resp = await client.get(
            f"http://payment-service/api/payments?order_id={order_id}",
            timeout=5.0
        )
        resp.raise_for_status()
        return resp.json()
    
    async def _get_shipping(self, client, order_id):
        resp = await client.get(
            f"http://shipping-service/api/shipping?order_id={order_id}",
            timeout=5.0
        )
        resp.raise_for_status()
        return resp.json()

# This reduces the client from 3 HTTP calls to 1
"""
    print(code)

response_aggregation()

# rate_limiting.py
# Rate limiting implementation

def rate_limiting():
    print("API Gateway Rate Limiting")
    print("=" * 40)
    print()
    
    code = """
import time
import redis

class RateLimiter:
    """Sliding window rate limiter."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def check_rate_limit(self, client_id, limit, window_seconds=60):
        """
        Check if client_id has exceeded the rate limit.
        Returns (allowed, remaining, reset_time).
        """
        key = f"ratelimit:{client_id}"
        now = time.time()
        window_start = now - window_seconds
        
        # Remove old entries
        self.redis.zremrangebyscore(key, 0, window_start)
        
        # Count requests in current window
        request_count = self.redis.zcard(key)
        
        if request_count >= limit:
            # Get oldest request time for reset header
            oldest = self.redis.zrange(key, 0, 0, withscores=True)
            reset_time = oldest[0][1] + window_seconds if oldest else now
            return False, 0, int(reset_time)
        
        # Add current request
        self.redis.zadd(key, {str(now): now})
        self.redis.expire(key, window_seconds)
        
        remaining = limit - request_count - 1
        return True, remaining, int(now + window_seconds)
    
    def middleware(self, request, limit=100, window=60):
        client_id = self._get_client_id(request)
        allowed, remaining, reset = self.check_rate_limit(
            client_id, limit, window
        )
        
        if not allowed:
            return {
                "error": "rate_limit_exceeded",
                "message": f"Limit of {limit} requests per {window}s exceeded",
                "retry_after": reset - time.time()
            }, 429
        
        # Add rate limit headers
        request.response_headers["X-RateLimit-Remaining"] = str(remaining)
        request.response_headers["X-RateLimit-Reset"] = str(reset)
        
        return None  # Continue processing
    
    def _get_client_id(self, request):
        # Use API key, user ID, or IP address
        return request.headers.get("X-API-Key") or \
               request.headers.get("X-User-Id") or \
               request.client.host
"""
    print(code)

rate_limiting()

def rideshare_gateway():
    print("Ride-Sharing API Gateway Design")
    print("=" * 45)
    print()
    print("Routes:")
    print("  POST /api/auth/login          -> user-service")
    print("  GET  /api/users/profile       -> user-service")
    print("  POST /api/rides/request       -> ride-service")
    print("  GET  /api/rides/{id}          -> ride-service")
    print("  GET  /api/rides/{id}/detail   -> AGGREGATED")
    print("  POST /api/payments/methods    -> payment-service")
    print("  GET  /api/drivers/nearby      -> driver-service")
    print()
    print("Aggregated Endpoint: GET /api/rides/{id}/detail")
    print("  Calls: ride-service (ride details)")
    print("         driver-service (driver name, rating, car)")
    print("         payment-service (fare breakdown)")
    print("  Returns: Combined JSON in single response")
    print()
    print("Rate Limiting:")
    print("  Free users:  20 requests/min")
    print("  Premium:    100 requests/min")
    print("  Drivers:    200 requests/min")

rideshare_gateway()