API Gateway Pattern

Introduction

An API Gateway serves as the single entry point for all client requests. From building microservices systems, I've found that without a gateway, clients must know about every service endpoint, handle authentication for each, and manage cross-cutting concerns themselves. The gateway pattern centralizes these responsibilities.

This article covers gateway design, routing, authentication, rate limiting, and implementation with FastAPI.

Why API Gateway?

Gateway Responsibilities

Responsibility	Description
Routing	Direct requests to appropriate backend services
Authentication	Validate tokens, API keys
Authorization	Check permissions before forwarding
Rate Limiting	Prevent abuse and ensure fair usage
Load Balancing	Distribute traffic across service instances
Caching	Cache responses to reduce backend load
Request Transformation	Modify requests before forwarding
Response Aggregation	Combine multiple service responses
Monitoring	Log requests, track metrics

Basic Gateway Implementation

from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import JSONResponse
import httpx
from typing import Optional
import logging

app = FastAPI(title="API Gateway")
logger = logging.getLogger(__name__)

# Service registry
SERVICE_REGISTRY = {
    "users": "http://user-service:8001",
    "orders": "http://order-service:8002",
    "products": "http://product-service:8003",
    "payments": "http://payment-service:8004",
}


class GatewayRouter:
    """Route requests to backend services."""
    
    def __init__(self):
        self.client = httpx.AsyncClient(timeout=30.0)
    
    async def route(
        self,
        service: str,
        path: str,
        request: Request,
    ) -> JSONResponse:
        """Forward request to backend service."""
        if service not in SERVICE_REGISTRY:
            raise HTTPException(404, f"Service '{service}' not found")
        
        base_url = SERVICE_REGISTRY[service]
        url = f"{base_url}{path}"
        
        # Forward headers (excluding hop-by-hop headers)
        headers = dict(request.headers)
        headers.pop("host", None)
        headers.pop("content-length", None)
        
        # Forward request
        try:
            body = await request.body()
            response = await self.client.request(
                method=request.method,
                url=url,
                headers=headers,
                params=dict(request.query_params),
                content=body if body else None,
            )
            
            return JSONResponse(
                content=response.json() if response.content else None,
                status_code=response.status_code,
                headers=dict(response.headers),
            )
        
        except httpx.ConnectError:
            raise HTTPException(503, f"Service '{service}' unavailable")
        except httpx.TimeoutException:
            raise HTTPException(504, f"Service '{service}' timeout")


router = GatewayRouter()


@app.api_route(
    "/{service}/{path:path}",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
)
async def gateway_route(service: str, path: str, request: Request):
    """Main gateway endpoint."""
    return await router.route(service, f"/{path}", request)

Authentication at the Gateway

JWT Validation

from fastapi import Security, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from dataclasses import dataclass
from datetime import datetime


security = HTTPBearer()

JWT_SECRET = "your-secret-key"
JWT_ALGORITHM = "HS256"


@dataclass
class TokenPayload:
    user_id: str
    email: str
    roles: list[str]
    exp: datetime
    iat: datetime


class JWTValidator:
    """Validate JWT tokens at gateway."""
    
    def __init__(self, secret: str, algorithm: str = "HS256"):
        self.secret = secret
        self.algorithm = algorithm
    
    def validate(self, token: str) -> TokenPayload:
        """Validate and decode JWT token."""
        try:
            payload = jwt.decode(
                token,
                self.secret,
                algorithms=[self.algorithm],
            )
            return TokenPayload(
                user_id=payload["sub"],
                email=payload.get("email", ""),
                roles=payload.get("roles", []),
                exp=datetime.fromtimestamp(payload["exp"]),
                iat=datetime.fromtimestamp(payload["iat"]),
            )
        except jwt.ExpiredSignatureError:
            raise HTTPException(401, "Token expired")
        except jwt.InvalidTokenError as e:
            raise HTTPException(401, f"Invalid token: {e}")


jwt_validator = JWTValidator(JWT_SECRET)


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Security(security),
) -> TokenPayload:
    """Dependency to extract and validate user from token."""
    return jwt_validator.validate(credentials.credentials)


# Protected routes
@app.api_route(
    "/api/{service}/{path:path}",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
)
async def protected_route(
    service: str,
    path: str,
    request: Request,
    user: TokenPayload = Depends(get_current_user),
):
    """Protected gateway endpoint - requires authentication."""
    # Add user info to forwarded headers
    headers = dict(request.headers)
    headers["X-User-ID"] = user.user_id
    headers["X-User-Email"] = user.email
    headers["X-User-Roles"] = ",".join(user.roles)
    
    return await router.route(service, f"/{path}", request)

API Key Authentication

from fastapi import Header


class APIKeyValidator:
    """Validate API keys for service-to-service auth."""
    
    def __init__(self, valid_keys: dict[str, dict]):
        # key -> {"client_id": ..., "scopes": [...], "rate_limit": ...}
        self.valid_keys = valid_keys
    
    def validate(self, api_key: str) -> dict:
        """Validate API key and return client info."""
        if api_key not in self.valid_keys:
            raise HTTPException(401, "Invalid API key")
        return self.valid_keys[api_key]


api_key_validator = APIKeyValidator({
    "key-12345": {
        "client_id": "mobile-app",
        "scopes": ["read:users", "write:orders"],
        "rate_limit": 1000,
    },
    "key-67890": {
        "client_id": "partner-api",
        "scopes": ["read:products"],
        "rate_limit": 100,
    },
})


async def validate_api_key(
    x_api_key: str = Header(..., alias="X-API-Key"),
) -> dict:
    """Dependency to validate API key."""
    return api_key_validator.validate(x_api_key)


@app.api_route("/partner/{service}/{path:path}", methods=["GET"])
async def partner_route(
    service: str,
    path: str,
    request: Request,
    client: dict = Depends(validate_api_key),
):
    """Partner API endpoint with API key auth."""
    # Check scopes
    required_scope = f"read:{service}"
    if required_scope not in client["scopes"]:
        raise HTTPException(403, f"Scope '{required_scope}' required")
    
    return await router.route(service, f"/{path}", request)

Rate Limiting

Token Bucket Algorithm

import asyncio
from dataclasses import dataclass
from datetime import datetime
import redis.asyncio as redis


@dataclass
class RateLimitConfig:
    requests_per_second: int
    burst_size: int


class TokenBucketRateLimiter:
    """Rate limiter using token bucket algorithm."""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    async def is_allowed(
        self,
        key: str,
        config: RateLimitConfig,
    ) -> tuple[bool, dict]:
        """Check if request is allowed."""
        now = datetime.utcnow().timestamp()
        bucket_key = f"ratelimit:{key}"
        
        # Lua script for atomic token bucket
        script = """
        local key = KEYS[1]
        local rate = tonumber(ARGV[1])
        local burst = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])
        
        local bucket = redis.call('HMGET', key, 'tokens', 'last_update')
        local tokens = tonumber(bucket[1]) or burst
        local last_update = tonumber(bucket[2]) or now
        
        -- Refill tokens
        local elapsed = now - last_update
        tokens = math.min(burst, tokens + elapsed * rate)
        
        -- Check if request allowed
        local allowed = 0
        if tokens >= 1 then
            tokens = tokens - 1
            allowed = 1
        end
        
        -- Update bucket
        redis.call('HMSET', key, 'tokens', tokens, 'last_update', now)
        redis.call('EXPIRE', key, 3600)
        
        return {allowed, tokens}
        """
        
        result = await self.redis.eval(
            script,
            1,
            bucket_key,
            config.requests_per_second,
            config.burst_size,
            now,
        )
        
        allowed = bool(result[0])
        remaining = int(result[1])
        
        return allowed, {
            "X-RateLimit-Limit": str(config.requests_per_second),
            "X-RateLimit-Remaining": str(max(0, remaining)),
            "X-RateLimit-Reset": str(int(now) + 1),
        }


# Rate limit middleware
rate_limiter = TokenBucketRateLimiter(redis.from_url("redis://localhost:6379"))


async def check_rate_limit(
    request: Request,
    user: TokenPayload = Depends(get_current_user),
) -> dict:
    """Rate limit by user ID."""
    config = RateLimitConfig(requests_per_second=10, burst_size=20)
    
    allowed, headers = await rate_limiter.is_allowed(
        f"user:{user.user_id}",
        config,
    )
    
    if not allowed:
        raise HTTPException(
            429,
            "Rate limit exceeded",
            headers=headers,
        )
    
    return headers


@app.api_route(
    "/api/v1/{service}/{path:path}",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
)
async def rate_limited_route(
    service: str,
    path: str,
    request: Request,
    user: TokenPayload = Depends(get_current_user),
    rate_limit_headers: dict = Depends(check_rate_limit),
):
    """Rate limited API endpoint."""
    response = await router.route(service, f"/{path}", request)
    response.headers.update(rate_limit_headers)
    return response

Tiered Rate Limiting

from enum import Enum


class PricingTier(Enum):
    FREE = "free"
    BASIC = "basic"
    PRO = "pro"
    ENTERPRISE = "enterprise"


TIER_LIMITS = {
    PricingTier.FREE: RateLimitConfig(requests_per_second=1, burst_size=5),
    PricingTier.BASIC: RateLimitConfig(requests_per_second=10, burst_size=20),
    PricingTier.PRO: RateLimitConfig(requests_per_second=50, burst_size=100),
    PricingTier.ENTERPRISE: RateLimitConfig(requests_per_second=500, burst_size=1000),
}


class TieredRateLimiter:
    """Rate limiter with different limits per tier."""
    
    def __init__(self, base_limiter: TokenBucketRateLimiter):
        self.limiter = base_limiter
    
    async def check(
        self,
        user_id: str,
        tier: PricingTier,
    ) -> tuple[bool, dict]:
        """Check rate limit based on user's tier."""
        config = TIER_LIMITS[tier]
        return await self.limiter.is_allowed(f"user:{user_id}", config)


async def get_user_tier(user: TokenPayload = Depends(get_current_user)) -> PricingTier:
    """Get user's pricing tier (from token or user service)."""
    # In practice, this might query user service or be in JWT
    tier_claim = getattr(user, "tier", "free")
    return PricingTier(tier_claim)

Request Aggregation

import asyncio
from dataclasses import dataclass


@dataclass
class AggregatedDashboard:
    user: dict
    recent_orders: list
    recommended_products: list


class ResponseAggregator:
    """Aggregate responses from multiple services."""
    
    def __init__(self, client: httpx.AsyncClient):
        self.client = client
    
    async def get_dashboard(self, user_id: str) -> AggregatedDashboard:
        """Fetch and aggregate dashboard data."""
        # Parallel requests
        user_task = self.client.get(
            f"{SERVICE_REGISTRY['users']}/users/{user_id}"
        )
        orders_task = self.client.get(
            f"{SERVICE_REGISTRY['orders']}/orders",
            params={"user_id": user_id, "limit": 5},
        )
        products_task = self.client.get(
            f"{SERVICE_REGISTRY['products']}/products/recommended",
            params={"user_id": user_id},
        )
        
        # Wait for all
        results = await asyncio.gather(
            user_task,
            orders_task,
            products_task,
            return_exceptions=True,
        )
        
        # Handle failures gracefully
        user_data = {}
        if not isinstance(results[0], Exception):
            user_data = results[0].json()
        
        orders_data = []
        if not isinstance(results[1], Exception):
            orders_data = results[1].json()
        
        products_data = []
        if not isinstance(results[2], Exception):
            products_data = results[2].json()
        
        return AggregatedDashboard(
            user=user_data,
            recent_orders=orders_data,
            recommended_products=products_data,
        )


aggregator = ResponseAggregator(httpx.AsyncClient(timeout=10.0))


@app.get("/api/dashboard")
async def get_dashboard(user: TokenPayload = Depends(get_current_user)):
    """Aggregated dashboard endpoint."""
    dashboard = await aggregator.get_dashboard(user.user_id)
    return {
        "user": dashboard.user,
        "recent_orders": dashboard.recent_orders,
        "recommended_products": dashboard.recommended_products,
    }

Caching at the Gateway

import hashlib
import json
from typing import Optional


class GatewayCache:
    """Cache responses at the gateway."""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    def _cache_key(self, request: Request) -> str:
        """Generate cache key from request."""
        key_parts = [
            request.method,
            str(request.url.path),
            str(sorted(request.query_params.items())),
        ]
        key_string = "|".join(key_parts)
        return f"cache:{hashlib.md5(key_string.encode()).hexdigest()}"
    
    async def get(self, request: Request) -> Optional[dict]:
        """Get cached response."""
        key = self._cache_key(request)
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None
    
    async def set(
        self,
        request: Request,
        response: dict,
        ttl: int = 60,
    ) -> None:
        """Cache response."""
        key = self._cache_key(request)
        await self.redis.setex(key, ttl, json.dumps(response))
    
    async def invalidate(self, pattern: str) -> None:
        """Invalidate cache by pattern."""
        keys = await self.redis.keys(f"cache:*{pattern}*")
        if keys:
            await self.redis.delete(*keys)


cache = GatewayCache(redis.from_url("redis://localhost:6379"))


# Cacheable endpoints configuration
CACHE_CONFIG = {
    ("GET", "/products"): 300,  # 5 minutes
    ("GET", "/categories"): 3600,  # 1 hour
    ("GET", "/products/{id}"): 60,  # 1 minute
}


async def cache_middleware(request: Request, call_next):
    """Caching middleware."""
    # Only cache GET requests
    if request.method != "GET":
        response = await call_next(request)
        # Invalidate cache on writes
        if request.method in ["POST", "PUT", "DELETE", "PATCH"]:
            await cache.invalidate(request.url.path.split("/")[1])
        return response
    
    # Check cache
    cached = await cache.get(request)
    if cached:
        return JSONResponse(
            content=cached["body"],
            status_code=cached["status"],
            headers={"X-Cache": "HIT"},
        )
    
    # Forward request
    response = await call_next(request)
    
    # Cache successful responses
    if 200 <= response.status_code < 300:
        body = b"".join([chunk async for chunk in response.body_iterator])
        ttl = CACHE_CONFIG.get((request.method, request.url.path), 60)
        
        await cache.set(
            request,
            {"body": json.loads(body), "status": response.status_code},
            ttl,
        )
        
        return JSONResponse(
            content=json.loads(body),
            status_code=response.status_code,
            headers={"X-Cache": "MISS"},
        )
    
    return response

Circuit Breaker in Gateway

from circuitbreaker import circuit
from datetime import timedelta


class ServiceCircuitBreaker:
    """Circuit breaker per service."""
    
    def __init__(self):
        self.breakers = {}
    
    def get_breaker(self, service: str):
        """Get or create circuit breaker for service."""
        if service not in self.breakers:
            @circuit(
                failure_threshold=5,
                recovery_timeout=30,
                expected_exception=Exception,
            )
            async def call_service(url: str, **kwargs):
                async with httpx.AsyncClient() as client:
                    return await client.request(**kwargs)
            
            self.breakers[service] = call_service
        
        return self.breakers[service]


service_breaker = ServiceCircuitBreaker()


class ResilientGatewayRouter:
    """Gateway router with circuit breaker."""
    
    async def route(
        self,
        service: str,
        path: str,
        request: Request,
    ) -> JSONResponse:
        """Route with circuit breaker protection."""
        if service not in SERVICE_REGISTRY:
            raise HTTPException(404, f"Service '{service}' not found")
        
        base_url = SERVICE_REGISTRY[service]
        url = f"{base_url}{path}"
        
        breaker = service_breaker.get_breaker(service)
        
        try:
            response = await breaker(
                url,
                method=request.method,
                headers=dict(request.headers),
                params=dict(request.query_params),
                content=await request.body(),
            )
            
            return JSONResponse(
                content=response.json(),
                status_code=response.status_code,
            )
        
        except Exception as e:
            # Circuit is open or service failed
            logger.error(f"Service {service} call failed: {e}")
            raise HTTPException(503, f"Service '{service}' temporarily unavailable")

Request Transformation

from pydantic import BaseModel
from typing import Any


class RequestTransformer:
    """Transform requests before forwarding."""
    
    def __init__(self):
        self.transformers = {}
    
    def register(self, service: str, path_pattern: str):
        """Decorator to register transformer."""
        def decorator(func):
            key = f"{service}:{path_pattern}"
            self.transformers[key] = func
            return func
        return decorator
    
    async def transform(
        self,
        service: str,
        path: str,
        request: Request,
    ) -> tuple[str, dict, Any]:
        """Transform request if transformer exists."""
        # Find matching transformer
        for pattern, transformer in self.transformers.items():
            svc, path_pattern = pattern.split(":", 1)
            if svc == service and self._matches(path, path_pattern):
                return await transformer(path, request)
        
        # No transformation
        return path, dict(request.headers), await request.body()


transformer = RequestTransformer()


@transformer.register("orders", "/orders")
async def transform_create_order(path: str, request: Request):
    """Transform order creation request."""
    body = await request.json()
    
    # Enrich with server-side data
    body["created_at"] = datetime.utcnow().isoformat()
    body["source"] = "api_gateway"
    
    # Add correlation ID
    headers = dict(request.headers)
    headers["X-Correlation-ID"] = str(uuid.uuid4())
    
    return path, headers, json.dumps(body).encode()


@transformer.register("users", "/users/*/profile")
async def transform_profile_request(path: str, request: Request):
    """Transform profile requests to include additional fields."""
    headers = dict(request.headers)
    headers["X-Include-Stats"] = "true"
    
    return path, headers, await request.body()

Health Checks and Monitoring

from datetime import datetime


class GatewayHealthChecker:
    """Monitor health of backend services."""
    
    def __init__(self, services: dict[str, str]):
        self.services = services
        self.health_status = {}
    
    async def check_service(self, name: str, url: str) -> dict:
        """Check single service health."""
        try:
            async with httpx.AsyncClient(timeout=5.0) as client:
                response = await client.get(f"{url}/health")
                
                return {
                    "name": name,
                    "status": "healthy" if response.status_code == 200 else "unhealthy",
                    "latency_ms": response.elapsed.total_seconds() * 1000,
                    "checked_at": datetime.utcnow().isoformat(),
                }
        except Exception as e:
            return {
                "name": name,
                "status": "unhealthy",
                "error": str(e),
                "checked_at": datetime.utcnow().isoformat(),
            }
    
    async def check_all(self) -> dict:
        """Check all services."""
        tasks = [
            self.check_service(name, url)
            for name, url in self.services.items()
        ]
        
        results = await asyncio.gather(*tasks)
        self.health_status = {r["name"]: r for r in results}
        
        all_healthy = all(
            r["status"] == "healthy" for r in results
        )
        
        return {
            "gateway_status": "healthy" if all_healthy else "degraded",
            "services": results,
            "checked_at": datetime.utcnow().isoformat(),
        }


health_checker = GatewayHealthChecker(SERVICE_REGISTRY)


@app.get("/health")
async def health():
    """Gateway health endpoint."""
    return await health_checker.check_all()


@app.get("/health/live")
async def liveness():
    """Kubernetes liveness probe."""
    return {"status": "alive"}


@app.get("/health/ready")
async def readiness():
    """Kubernetes readiness probe."""
    health = await health_checker.check_all()
    if health["gateway_status"] != "healthy":
        raise HTTPException(503, "Not ready")
    return {"status": "ready"}

Complete Gateway Example

from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import uvicorn


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await router.client.__aenter__()
    yield
    # Shutdown
    await router.client.__aexit__(None, None, None)


app = FastAPI(
    title="API Gateway",
    description="Unified entry point for microservices",
    lifespan=lifespan,
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


# Public routes (no auth)
@app.api_route(
    "/public/{service}/{path:path}",
    methods=["GET"],
)
async def public_route(service: str, path: str, request: Request):
    """Public endpoints - no authentication required."""
    if service not in ["products", "categories"]:
        raise HTTPException(403, "Service not publicly accessible")
    return await router.route(service, f"/{path}", request)


# Authenticated routes
@app.api_route(
    "/api/v1/{service}/{path:path}",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
)
async def authenticated_route(
    service: str,
    path: str,
    request: Request,
    user: TokenPayload = Depends(get_current_user),
    _: dict = Depends(check_rate_limit),
):
    """Authenticated API endpoints."""
    return await router.route(service, f"/{path}", request)


# Aggregated endpoints
@app.get("/api/v1/me/dashboard")
async def my_dashboard(user: TokenPayload = Depends(get_current_user)):
    """User's aggregated dashboard."""
    return await aggregator.get_dashboard(user.user_id)


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Key Takeaways

1. Single entry point - Simplifies client integration

2. Authentication centralization - Validate once at the edge

3. Rate limiting - Protect backend services from abuse

4. Response aggregation - Reduce client round trips

5. Circuit breakers - Prevent cascade failures

What's Next?

With the gateway handling cross-cutting concerns, each service needs its own data. In Article 7: Database per Service, we'll explore data ownership and polyglot persistence strategies.

---

This article is part of the Microservice Architecture 101 series.