Introduction to API Gateway

# gateway_concept.py
# Illustrating the core concept of an API gateway
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Optional

class RouteMethod(Enum):
    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"

@dataclass
class Route:
    path: str
    methods: List[RouteMethod]
    target_service: str
    target_path: str

@dataclass
class GatewayConfig:
    routes: Dict[str, Route]
    rate_limit_per_minute: int
    enable_auth: bool

class APIGateway:
    """Simple API gateway demonstrating core routing."""

    def __init__(self, config: GatewayConfig):
        self.config = config
        self.request_log = []

    def route_request(self, method: str, path: str, headers: Dict) -> dict:
        """Route an incoming request to the appropriate service."""
        matched_route = None
        for route_path, route in self.config.routes.items():
            if path.startswith(route_path):
                matched_route = route
                break

        if not matched_route:
            return {"status": 404, "body": {"error": "Route not found"}}

        route_method = RouteMethod(method)
        if route_method not in matched_route.methods:
            return {"status": 405, "body": {"error": "Method not allowed"}}

        self.request_log.append({
            "path": path,
            "method": method,
            "target": matched_route.target_service,
        })

        return {
            "status": 200,
            "body": {
                "message": f"Routing to {matched_route.target_service}",
                "service": matched_route.target_service,
                "path": matched_route.target_path,
            }
        }

config = GatewayConfig(
    routes={
        "/products": Route(
            path="/products",
            methods=[RouteMethod.GET, RouteMethod.POST],
            target_service="product-service",
            target_path="/api/products",
        ),
        "/orders": Route(
            path="/orders",
            methods=[RouteMethod.GET, RouteMethod.POST],
            target_service="order-service",
            target_path="/api/orders",
        ),
        "/auth": Route(
            path="/auth",
            methods=[RouteMethod.POST],
            target_service="auth-service",
            target_path="/api/auth/login",
        ),
    },
    rate_limit_per_minute=100,
    enable_auth=True,
)

gateway = APIGateway(config)

result1 = gateway.route_request("GET", "/products", {})
print(f"Result 1: {result1}")

result2 = gateway.route_request("POST", "/auth", {})
print(f"Result 2: {result2}")

result3 = gateway.route_request("DELETE", "/products", {})
print(f"Result 3: {result3}")

print(f"\nLogged requests: {len(gateway.request_log)}")

# gateway_simulator.py
import time
import json
from typing import Dict, List, Optional

class GatewaySimulator:
    def __init__(self):
        self.services = {}
        self.routes = {}
        self.logs = []

    def register_service(self, name: str, base_url: str):
        self.services[name] = {"base_url": base_url, "healthy": True}

    def add_route(self, path: str, service: str):
        self.routes[path] = service

    def process_request(self, method: str, path: str) -> dict:
        start = time.time()
        self.logs.append({"method": method, "path": path, "timestamp": start})

        matched_service = None
        for route_path, service_name in self.routes.items():
            if path.startswith(route_path):
                matched_service = service_name
                break

        if not matched_service:
            return {"status": 404, "body": "Route not found", "latency_ms": 0}

        if not self.services.get(matched_service, {}).get("healthy", False):
            return {"status": 503, "body": "Service unavailable", "latency_ms": 0}

        latency_ms = (time.time() - start) * 1000
        return {
            "status": 200,
            "body": f"Routed to {matched_service}",
            "target": self.services[matched_service]["base_url"],
            "latency_ms": round(latency_ms, 2),
        }

sim = GatewaySimulator()
sim.register_service("products", "http://products:3000")
sim.register_service("orders", "http://orders:4000")
sim.add_route("/api/products", "products")
sim.add_route("/api/orders", "orders")

for path in ["/api/products", "/api/orders", "/api/users"]:
    result = sim.process_request("GET", path)
    print(f"{path}: {result['status']} -> {result['body']}")