API gateways sit between clients and backend services, providing a single entry point with cross-cutting concerns: authentication, rate limiting, request transformation, and routing.
API Gateway Fundamentals
Architecture Pattern
┌─────────────────┐
│ Clients │
│ (Web, Mobile) │
└────────┬────────┘
│
┌────────▼──────────┐
│ API Gateway │
│ - Routing │
│ - Auth/Rate Limit│
│ - Transform │
│ - Load Balance │
└────────┬──────────┘
│
┌────────────────────┼────────────────────┐
│ │ │
┌────▼──────┐ ┌──────▼──────┐ ┌─────▼──────┐
│ API v1 │ │ API v2 │ │ Internal │
│ Service │ │ Service │ │ Service │
└───────────┘ └─────────────┘ └────────────┘
Core Responsibilities
class APIGateway:
"""Central API Gateway responsibilities"""
def __init__(self):
self.responsibilities = {
'routing': self._route_requests,
'authentication': self._authenticate,
'rate_limiting': self._enforce_rate_limits,
'transformation': self._transform_request,
'load_balancing': self._balance_load,
'caching': self._cache_responses,
'logging': self._log_requests,
'circuit_breaking': self._handle_failures
}
def _route_requests(self, request):
"""Route requests to appropriate backend service"""
# Parse URL path and route to service
if request.path.startswith('/api/v1/users'):
return 'user-service:8080'
elif request.path.startswith('/api/v1/orders'):
return 'order-service:8080'
else:
return 'fallback-service:8080'
def _authenticate(self, request):
"""Verify API key, JWT, or OAuth token"""
auth_header = request.headers.get('Authorization')
if not auth_header:
raise UnauthorizedError('Missing auth header')
# Validate JWT
token = auth_header.replace('Bearer ', '')
try:
payload = jwt.decode(token, 'secret', algorithms=['HS256'])
return payload['user_id']
except jwt.InvalidTokenError:
raise UnauthorizedError('Invalid token')
def _enforce_rate_limits(self, client_id: str):
"""Rate limiting: allow N requests per minute"""
key = f'rate_limit:{client_id}'
current = redis.incr(key)
if current == 1:
redis.expire(key, 60) # 60 second window
if current > 100: # 100 req/min limit
raise RateLimitError(f'Rate limit exceeded')
def _transform_request(self, request):
"""Transform request before sending to backend"""
# Add correlation ID
request.headers['X-Correlation-Id'] = uuid.uuid4()
# Add API version header
request.headers['X-API-Version'] = '1.0'
# Add timestamp
request.headers['X-Request-Time'] = datetime.utcnow().isoformat()
return request
def _balance_load(self, service_name: str):
"""Distribute requests across service instances"""
instances = [
'service-1.internal:8080',
'service-2.internal:8080',
'service-3.internal:8080'
]
# Round-robin selection
index = hash(service_name) % len(instances)
return instances[index]
Kong API Gateway
Architecture & Setup
## Kong deployment with PostgreSQL
apiVersion: v1
kind: Service
metadata:
name: kong-admin
spec:
ports:
- port: 8001
name: admin
selector:
app: kong
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: kong
spec:
replicas: 3
selector:
matchLabels:
app: kong
template:
metadata:
labels:
app: kong
spec:
containers:
- name: kong
image: kong:3.0
ports:
- containerPort: 8000 # proxy
- containerPort: 8001 # admin
- containerPort: 8443 # proxy ssl
env:
- name: KONG_DATABASE
value: postgres
- name: KONG_PG_HOST
value: postgres
- name: KONG_PROXY_ACCESS_LOG
value: /dev/stdout
- name: KONG_ADMIN_ACCESS_LOG
value: /dev/stdout
livenessProbe:
httpGet:
path: /status
port: 8001
initialDelaySeconds: 30
Kong Configuration via API
import requests
import json
class KongGateway:
"""Configure Kong API Gateway"""
def __init__(self, admin_url='http://localhost:8001'):
self.admin_url = admin_url
def add_service(self, name: str, url: str):
"""Register backend service"""
response = requests.post(
f'{self.admin_url}/services',
json={
'name': name,
'url': url,
'protocol': 'http',
'host': url.split('://')[1].split('/')[0],
'port': 80,
'path': '/api'
}
)
return response.json()
def add_route(self, service_name: str, paths: list):
"""Create routes to service"""
response = requests.post(
f'{self.admin_url}/services/{service_name}/routes',
json={
'paths': paths,
'methods': ['GET', 'POST', 'PUT', 'DELETE'],
'protocols': ['http', 'https']
}
)
return response.json()
def enable_rate_limiting(self, service_name: str, limit: int):
"""Enable rate limiting plugin"""
response = requests.post(
f'{self.admin_url}/services/{service_name}/plugins',
json={
'name': 'rate-limiting',
'config': {
'minute': limit,
'policy': 'redis',
'redis_host': 'redis',
'redis_port': 6379
}
}
)
return response.json()
def enable_jwt_auth(self, service_name: str):
"""Enable JWT authentication"""
response = requests.post(
f'{self.admin_url}/services/{service_name}/plugins',
json={
'name': 'jwt',
'config': {
'secret_is_base64': False,
'key_claim_name': 'iss',
'claims_to_verify': ['exp']
}
}
)
return response.json()
def enable_cors(self, service_name: str):
"""Enable CORS headers"""
response = requests.post(
f'{self.admin_url}/services/{service_name}/plugins',
json={
'name': 'cors',
'config': {
'origins': ['*'],
'methods': ['GET', 'POST', 'PUT', 'DELETE'],
'headers': ['Accept', 'Content-Type', 'Authorization'],
'exposed_headers': ['X-Total-Count'],
'credentials': True,
'max_age': 3600
}
}
)
return response.json()
def add_response_transformer(self, service_name: str):
"""Transform API responses"""
response = requests.post(
f'{self.admin_url}/services/{service_name}/plugins',
json={
'name': 'response-transformer',
'config': {
'add': {
'headers': ['X-Gateway-Version:1.0'],
'json': ['response_time:${requestTime}']
}
}
}
)
return response.json()
## Usage
kong = KongGateway()
kong.add_service('user-service', 'http://user-service:8080')
kong.add_route('user-service', ['/api/users', '/api/users/.*'])
kong.enable_rate_limiting('user-service', 1000) # 1000 req/min
kong.enable_jwt_auth('user-service')
AWS API Gateway
REST API vs HTTP API
import boto3
import json
class AWSAPIGateway:
"""AWS API Gateway configuration"""
def __init__(self, region='us-east-1'):
self.client = boto3.client('apigateway', region_name=region)
self.lambda_client = boto3.client('lambda', region_name=region)
def create_rest_api(self, name: str, description: str):
"""Create REST API"""
response = self.client.create_rest_api(
name=name,
description=description,
endpointConfiguration={'types': ['REGIONAL']},
policy={
'Version': '2012-10-17',
'Statement': [{
'Effect': 'Allow',
'Principal': '*',
'Action': 'execute-api:Invoke',
'Resource': 'arn:aws:execute-api:*:*:*'
}]
}
)
return response['id']
def add_resource(self, api_id: str, parent_id: str, path_part: str):
"""Create API resource"""
response = self.client.create_resource(
restApiId=api_id,
parentId=parent_id,
pathPart=path_part
)
return response['id']
def create_method_with_lambda(self, api_id: str, resource_id: str,
lambda_arn: str):
"""Connect Lambda function as method"""
# Create POST method
self.client.put_method(
restApiId=api_id,
resourceId=resource_id,
httpMethod='POST',
type='AWS_PROXY', # Full request/response proxy
authorizationType='AWS_IAM'
)
# Integration with Lambda
self.client.put_integration(
restApiId=api_id,
resourceId=resource_id,
httpMethod='POST',
type='AWS_PROXY',
integrationHttpMethod='POST',
uri=lambda_arn
)
def enable_api_key_auth(self, api_id: str, resource_id: str):
"""Require API key for access"""
# Create API key
key_response = self.client.create_api_key(
name='mobile-app-key',
enabled=True
)
api_key_id = key_response['id']
# Create usage plan
plan_response = self.client.create_usage_plan(
name='mobile-app-plan',
description='Rate limited plan for mobile app',
apiStages=[{
'apiId': api_id,
'stage': 'prod'
}],
throttle={
'rateLimit': 1000, # 1000 req/sec
'burstLimit': 2000 # 2000 peak
},
quota={
'limit': 1000000, # 1M req/day
'period': 'DAY'
}
)
# Associate key with plan
self.client.create_usage_plan_key(
usagePlanId=plan_response['id'],
keyId=api_key_id,
keyType='API_KEY'
)
return api_key_id
def add_request_validator(self, api_id: str):
"""Validate request body and parameters"""
response = self.client.create_request_validator(
restApiId=api_id,
name='all-validator',
validateRequestBody=True,
validateRequestParameters=True
)
return response['id']
def add_cors_headers(self, api_id: str, resource_id: str):
"""Enable CORS for browser clients"""
self.client.put_method(
restApiId=api_id,
resourceId=resource_id,
httpMethod='OPTIONS',
authorizationType='NONE'
)
self.client.put_integration(
restApiId=api_id,
resourceId=resource_id,
httpMethod='OPTIONS',
type='MOCK',
requestTemplates={'application/json': '{"statusCode": 200}'}
)
self.client.put_integration_response(
restApiId=api_id,
resourceId=resource_id,
httpMethod='OPTIONS',
statusCode='200',
responseParameters={
'method.response.header.Access-Control-Allow-Headers':
'"Content-Type,X-Amz-Date,Authorization,X-Api-Key"',
'method.response.header.Access-Control-Allow-Methods':
'"*"',
'method.response.header.Access-Control-Allow-Origin':
'"*"'
}
)
## Usage
api = AWSAPIGateway()
api_id = api.create_rest_api('mobile-backend', 'Mobile app backend')
resource_id = api.add_resource(api_id, 'root_id', 'users')
api.create_method_with_lambda(api_id, resource_id, 'arn:aws:lambda:...')
api_key = api.enable_api_key_auth(api_id, resource_id)
Request Transformation Patterns
Header Manipulation
## Nginx API Gateway configuration
upstream backend {
server api-service-1:8080;
server api-service-2:8080;
server api-service-3:8080;
}
server {
listen 80;
server_name api.example.com;
# Add security headers
add_header X-Content-Type-Options nosniff;
add_header X-Frame-Options DENY;
add_header X-XSS-Protection "1; mode=block";
# Add custom headers
add_header X-API-Gateway "nginx/1.0";
add_header X-Request-ID $request_id;
location / {
# Add upstream headers
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# Add correlation ID
proxy_set_header X-Correlation-ID $request_id;
# Add timestamp
proxy_set_header X-Request-Time $msec;
proxy_pass http://backend;
# Timeouts
proxy_connect_timeout 5s;
proxy_send_timeout 10s;
proxy_read_timeout 10s;
}
}
Request Body Transformation
class RequestTransformer:
"""Transform API requests before forwarding"""
def transform_incoming(self, request):
"""Normalize incoming request"""
# Add metadata
request['_gateway_received_at'] = datetime.utcnow().isoformat()
request['_request_id'] = str(uuid.uuid4())
request['_client_ip'] = request.get('remote_addr')
# Normalize JSON paths (snake_case to camelCase)
if request['content_type'] == 'application/json':
request['body'] = self._normalize_json(request['body'])
return request
def transform_outgoing(self, response, request):
"""Add response metadata"""
response['_request_duration_ms'] = (
time.time() - request['_start_time']
) * 1000
response['_request_id'] = request['_request_id']
response['_api_version'] = '1.0'
return response
def _normalize_json(self, data):
"""Convert snake_case to camelCase"""
if isinstance(data, dict):
return {
self._to_camel_case(k): self._normalize_json(v)
for k, v in data.items()
}
elif isinstance(data, list):
return [self._normalize_json(item) for item in data]
return data
def _to_camel_case(self, snake_str):
components = snake_str.split('_')
return components[0] + ''.join(x.title() for x in components[1:])
API Gateway Comparison
class APIGatewayComparison:
"""Compare API gateway solutions"""
comparison = {
'Kong': {
'architecture': 'Nginx-based, self-hosted',
'scalability': '1000s of req/sec per node',
'latency': '1-5ms overhead',
'plugins': '30+ native plugins',
'cost': '$0 (open source) - $10k/year (enterprise)',
'best_for': 'High-volume, full control needed',
'deployment': 'Docker, Kubernetes, bare metal',
'features': {
'auth': ['OAuth2', 'JWT', 'API Key', 'LDAP'],
'ratelimit': 'Redis-backed, distributed',
'logging': 'Syslog, UDP, HTTP',
'analytics': 'Built-in request logging'
}
},
'AWS_API_Gateway': {
'architecture': 'Fully managed AWS service',
'scalability': 'Automatic, unlimited',
'latency': '10-50ms (regional)',
'plugins': 'Limited, AWS ecosystem focused',
'cost': '$3.50/million requests + cache/data',
'best_for': 'AWS-native, serverless',
'deployment': 'Managed by AWS',
'features': {
'auth': ['IAM', 'Cognito', 'Lambda authorizers'],
'ratelimit': 'Built-in throttling',
'logging': 'CloudWatch, S3',
'analytics': 'CloudWatch metrics'
}
},
'Nginx': {
'architecture': 'Lightweight, simple proxy',
'scalability': '100k+ req/sec single server',
'latency': '<1ms overhead',
'plugins': 'Community modules available',
'cost': '$0 (open source)',
'best_for': 'Simple routing, high performance',
'deployment': 'Docker, systemd, Kubernetes',
'features': {
'auth': 'Basic, JWT (with modules)',
'ratelimit': 'Built-in limit_req module',
'logging': 'Local files, syslog',
'analytics': 'Manual log parsing'
}
},
'Traefik': {
'architecture': 'Cloud-native, container-aware',
'scalability': 'Automatic with Kubernetes',
'latency': '1-3ms overhead',
'plugins': 'Growing ecosystem',
'cost': '$0 (open source)',
'best_for': 'Kubernetes-first, dynamic routing',
'deployment': 'Kubernetes, Docker Compose',
'features': {
'auth': ['Basic', 'Digest', 'OAuth2'],
'ratelimit': 'Built-in middleware',
'logging': 'Structured JSON logging',
'analytics': 'Metrics export'
}
}
}
## Decision Tree
```python
def choose_gateway(requirements):
"""Select appropriate API gateway"""
if requirements.get('serverless'):
return 'AWS_API_Gateway'
if requirements.get('kubernetes'):
return 'Traefik' if requirements.get('dynamic') else 'Nginx Ingress'
if requirements.get('full_control'):
return 'Kong' if requirements.get('features') else 'Nginx'
return 'AWS_API_Gateway'
Glossary
- API Gateway: Single entry point for backend services
- Request routing: Direct requests to appropriate backend
- Rate limiting: Control request frequency
- Request transformation: Modify headers/body before forwarding
- Circuit breaker: Fail fast when backend unavailable
- Load balancing: Distribute traffic across instances
Conclusion
API gateways are essential for managing cross-cutting concerns in microservice architectures. Kong offers the richest plugin ecosystem for complex requirements. AWS API Gateway is the natural choice for serverless and AWS-native stacks. Nginx provides the lowest latency for simple routing. Traefik excels in Kubernetes environments with automatic service discovery.
Choose the gateway that matches your deployment model and feature requirements. Start simple and add capabilities as needed—a basic reverse proxy may be sufficient before you need Kong’s plugin system or Traefik’s automatic discovery.
Resources
- Kong Documentation
- AWS API Gateway Guide
- Nginx Reverse Proxy
- Traefik Documentation
- API Gateway Patterns
Comments