Instruction file imported from cdalsoniii/brightpath-coder (.cursor/rules/008-api-gateway-patterns.mdc). Copyright stays with the author.

API Gateway Patterns

Core Responsibilities

Request Routing

routes:
  - path: /api/users/*
    service: user-service
    port: 8080
  
  - path: /api/orders/*
    service: order-service
    port: 8081
  
  - path: /api/products/*
    service: product-service
    port: 8082

Authentication/Authorization

@gateway.middleware
async def auth_middleware(request, call_next):
    token = request.headers.get("Authorization")
    
    # Validate token
    claims = await auth_service.validate(token)
    if not claims:
        return Response(status=401)
    
    # Check permissions
    if not has_permission(claims, request.path, request.method):
        return Response(status=403)
    
    # Add user context
    request.state.user = claims
    return await call_next(request)

Rate Limiting

# Token bucket implementation
@gateway.middleware
async def rate_limit(request, call_next):
    client_id = request.headers.get("X-Client-ID")
    
    # Check rate limit
    allowed = await rate_limiter.allow(
        key=client_id,
        limit=100,  # requests
        window=60   # seconds
    )
    
    if not allowed:
        return Response(
            status=429,
            headers={"Retry-After": "60"}
        )
    
    return await call_next(request)

Request Transformation

# Transform request before forwarding
async def transform_request(request):
    # Add correlation ID
    request.headers["X-Correlation-ID"] = str(uuid.uuid4())
    
    # Add timestamp
    request.headers["X-Request-Time"] = datetime.utcnow().isoformat()
    
    # Transform body if needed
    if request.path.startswith("/api/v1/"):
        request.body = transform_v1_to_v2(request.body)
    
    return request

Response Aggregation

# Combine responses from multiple services
async def get_order_details(order_id: str):
    # Parallel calls
    order, customer, products = await asyncio.gather(
        order_service.get(order_id),
        customer_service.get(order.customer_id),
        product_service.get_many(order.product_ids)
    )
    
    # Aggregate response
    return {
        "order": order,
        "customer": customer,
        "products": products
    }

Gateway Patterns

Backend for Frontend (BFF)

Mobile App → Mobile BFF → Microservices
Web App   → Web BFF    → Microservices
Admin     → Admin BFF  → Microservices

Edge Gateway

External Traffic → Edge Gateway → Internal Gateway → Services
                        ↓
              - SSL termination
              - DDoS protection
              - WAF

Circuit Breaker at Gateway

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.state = "CLOSED"
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = None
    
    async def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if self._should_try_reset():
                self.state = "HALF_OPEN"
            else:
                raise CircuitOpenError()
        
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

Gateway Security

IP Whitelisting

security:
  ip_whitelist:
    - 10.0.0.0/8
    - 192.168.0.0/16
  
  blocked_countries:
    - XX
    - YY

Request Validation

@gateway.middleware
async def validate_request(request, call_next):
    # Size limits
    if request.content_length > MAX_BODY_SIZE:
        return Response(status=413)
    
    # Content type validation
    if request.method in ["POST", "PUT"]:
        if "application/json" not in request.content_type:
            return Response(status=415)
    
    # Schema validation
    schema = get_schema(request.path, request.method)
    if schema and not validate(request.body, schema):
        return Response(status=400)
    
    return await call_next(request)

Observability

Request Tracing

@gateway.middleware
async def tracing(request, call_next):
    # Extract or create trace ID
    trace_id = request.headers.get("X-Trace-ID", str(uuid.uuid4()))
    span_id = str(uuid.uuid4())
    
    # Propagate headers
    request.headers["X-Trace-ID"] = trace_id
    request.headers["X-Span-ID"] = span_id
    
    with tracer.span(name=f"{request.method} {request.path}"):
        response = await call_next(request)
    
    # Add to response
    response.headers["X-Trace-ID"] = trace_id
    return response