API Gateway Mini Project — Complete Guide
In this tutorial, you'll build a complete API gateway simulation that combines routing, authentication, rate limiting, caching, response transformation, and monitoring.
What You'll Learn
By the end of this project, you will build a working API gateway simulation that handles real-world scenarios including authenticated routing, rate limiting, and fault tolerance.
Why It Matters
Building a gateway simulation gives you hands-on experience with the core concepts before implementing them with production tools like Kong or Envoy.
Real-World Use
This project simulates a production gateway managing three Microservices with authentication, rate limiting, caching, and health monitoring.
Gateway Architecture
flowchart TD
Client --> GW[API Gateway Simulation]
GW --> Auth[Auth Layer]
Auth --> Rate[Rate Limiter]
Rate --> Cache[Cache Layer]
Cache --> Router[Request Router]
Router --> US[User Service]
Router --> PS[Product Service]
Router --> OS[Order Service]
GW --> Monitor[Metrics & Logging]
GW --> Health[Health Endpoint]
Complete Gateway Implementation
# complete_gateway.py
import json
import time
import hashlib
from typing import Any, Dict, List, Optional, Tuple
from collections import defaultdict
class UserService:
def handle(self, method: str, path: str, body: Any = None) -> Dict:
if method == "GET" and path == "/users/me":
return {"status": 200, "body": {"id": 1, "name": "Alice", "role": "admin"}}
if method == "GET":
return {"status": 200, "body": {"users": [{"id": 1, "name": "Alice"}]}}
if method == "POST":
return {"status": 201, "body": {"created": True, "id": 2}}
return {"status": 404, "body": {"error": "Not found"}}
class ProductService:
def handle(self, method: str, path: str, body: Any = None) -> Dict:
if method == "GET":
return {"status": 200, "body": {"products": [{"id": 1, "name": "Widget", "price": 9.99}]}}
return {"status": 404, "body": {"error": "Not found"}}
class OrderService:
def handle(self, method: str, path: str, body: Any = None) -> Dict:
if method == "POST":
return {"status": 201, "body": {"order_id": 101, "total": 29.99}}
if method == "GET":
return {"status": 200, "body": {"orders": [{"id": 101, "total": 29.99}]}}
return {"status": 404, "body": {"error": "Not found"}}
class RateLimiter:
def __init__(self, max_requests: int = 10, window: int = 60):
self.max_requests = max_requests
self.window = window
self.clients: Dict[str, list] = defaultdict(list)
def check(self, client_id: str) -> Tuple[bool, int]:
now = time.time()
cutoff = now - self.window
self.clients[client_id] = [t for t in self.clients[client_id] if t > cutoff]
if len(self.clients[client_id]) >= self.max_requests:
return False, 0
self.clients[client_id].append(now)
remaining = self.max_requests - len(self.clients[client_id])
return True, remaining
class AuthMiddleware:
def __init__(self):
self.tokens = {"valid_token_123": {"user_id": 1, "role": "admin"}}
def authenticate(self, headers: Dict) -> Tuple[bool, Optional[Dict]]:
auth = headers.get("Authorization", "")
token = auth.replace("Bearer ", "")
user = self.tokens.get(token)
if not user:
return False, None
return True, user
class ResponseCache:
def __init__(self, ttl: int = 30):
self.ttl = ttl
self.cache: Dict[str, dict] = {}
def get(self, key: str) -> Optional[Dict]:
entry = self.cache.get(key)
if entry and time.time() < entry["expires"]:
return entry["data"]
self.cache.pop(key, None)
return None
def set(self, key: str, data: Dict):
self.cache[key] = {"data": data, "expires": time.time() + self.ttl}
class Monitor:
def __init__(self):
self.logs = []
self.request_count = 0
self.error_count = 0
def record(self, method: str, path: str, status: int, latency_ms: float):
self.request_count += 1
if status >= 500:
self.error_count += 1
self.logs.append({
"method": method, "path": path, "status": status,
"latency_ms": round(latency_ms, 2),
"timestamp": time.strftime("%H:%M:%S"),
})
def health(self) -> Dict:
return {
"status": "healthy",
"uptime_seconds": int(time.time() - self.start_time),
"total_requests": self.request_count,
"error_rate": round((self.error_count / max(1, self.request_count)) * 100, 2),
}
class APIGatewaySimulator:
def __init__(self):
self.services = {
"users": UserService(),
"products": ProductService(),
"orders": OrderService(),
}
self.routes = {
"/api/users": "users",
"/api/products": "products",
"/api/orders": "orders",
}
self.auth = AuthMiddleware()
self.rate_limiter = RateLimiter(max_requests=10, window=60)
self.cache = ResponseCache(ttl=30)
self.monitor = Monitor()
Monitor.start_time = time.time()
def handle(self, method: str, path: str, headers: Dict, body: Any = None) -> Dict:
start = time.time()
if path == "/health":
return {"status": 200, "body": self.monitor.health()}
authed, user = self.auth.authenticate(headers)
if not authed:
latency = (time.time() - start) * 1000
self.monitor.record(method, path, 401, latency)
return {"status": 401, "body": {"error": "Unauthorized"}}
client_id = str(user["user_id"])
allowed, remaining = self.rate_limiter.check(client_id)
if not allowed:
latency = (time.time() - start) * 1000
self.monitor.record(method, path, 429, latency)
return {"status": 429, "body": {"error": "Rate limit exceeded", "retry_after": 60}}
cache_key = f"{method}:{path}"
if method == "GET":
cached = self.cache.get(cache_key)
if cached:
latency = (time.time() - start) * 1000
self.monitor.record(method, path, 200, latency)
return {"status": 200, "body": cached, "source": "cache"}
matched_service = None
for prefix, service_name in sorted(self.routes.items(), key=lambda x: -len(x[0])):
if path.startswith(prefix):
matched_service = self.services[service_name]
break
if not matched_service:
latency = (time.time() - start) * 1000
self.monitor.record(method, path, 404, latency)
return {"status": 404, "body": {"error": "Route not found"}}
result = matched_service.handle(method, path, body)
if method == "GET" and result["status"] == 200:
self.cache.set(cache_key, result["body"])
latency = (time.time() - start) * 1000
self.monitor.record(method, path, result["status"], latency)
result["body"]["_user"] = user["role"]
return {"status": result["status"], "body": result["body"]}
gw = APIGatewaySimulator()
valid_token = "valid_token_123"
tests = [
("GET", "/api/users", {"Authorization": f"Bearer {valid_token}"}),
("GET", "/api/products", {"Authorization": f"Bearer {valid_token}"}),
("POST", "/api/orders", {"Authorization": f"Bearer {valid_token}"}, {"product_id": 1}),
("GET", "/api/users", {}),
("GET", "/health", {}),
]
for test in tests:
method, path, headers = test[0], test[1], test[2]
body = test[3] if len(test) > 3 else None
result = gw.handle(method, path, headers, body)
print(f"{method:5s} {path:20s} -> {result['status']} {result['body'].get('error', 'OK')}")
print(f"\nMonitoring:")
print(f" Requests: {gw.monitor.request_count}")
print(f" Error rate: {gw.monitor.health()['error_rate']}%")
Expected output:
GET /api/users -> 200 OK
GET /api/products -> 200 OK
POST /api/orders -> 201 OK
GET /api/users -> 401 Unauthorized
GET /health -> 200 OK
Monitoring:
Requests: 5
Error rate: 0.0%
Project Extension Ideas
- Add API key authentication alongside JWT
- Implement circuit breaker for failing services
- Add request logging to a file
- Support multiple rate limit tiers (free/pro/enterprise)
- Add response transformation to wrap responses in an envelope
- Implement Websocket upgrade support
- Add distributed rate limiting with a Redis simulation
Common Mistakes
1. Not Testing Error Paths
Test authentication failures, rate limit exceeded, service unavailable, and route not found scenarios.
2. Ignoring Latency Tracking
Every gateway operation adds latency. Measure and optimize the slowest parts of the pipeline.
3. Mixing Service Logic into Gateway
Keep the gateway focused on routing and cross-cutting concerns. Service logic belongs in the services.
4. No Graceful Degradation
When a service fails, return cached data or a helpful error instead of crashing the gateway.
5. Forgetting to Validate Configuration
Invalid route configurations can crash the gateway. Validate all configuration at startup.
Practice Questions
1. How does this gateway handle authentication?
It validates the Bearer token from the Authorization header against a pre-defined set of valid tokens, returning 401 if invalid.
2. How does caching improve performance?
Cached GET responses are returned instantly without forwarding to backend services, reducing latency from ~50ms to ~1ms.
3. What happens when rate limit is exceeded?
The gateway returns 429 Too Many Requests with a Retry-After header indicating when the client can retry.
4. How could you extend this to support WebSockets?
Add an upgrade check in the handle method that detects Upgrade: websocket headers and establishes a persistent connection.
Challenge
Extend the gateway to support WebSocket upgrade, add a circuit breaker for the order service, implement IP whitelisting for the admin routes, and expose Prometheus-style metrics at a /metrics endpoint.
FAQ
What's Next
Congratulations on completing the API Gateway series. Next, explore API versioning, then learn about GraphQL vs REST.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro