Caching at the Gateway — Deep Dive into Request and Response Caching
In this tutorial, you'll learn about Caching Deep. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Caching at the API Gateway reduces backend load by serving repeated requests from a fast cache layer, drastically improving response times and reducing infrastructure costs.
What You'll Learn
By the end of this lesson, you will implement gateway response caching, configure cache keys and TTLs, handle invalidation, use stale-while-revalidate, and integrate with Redis for distributed caching.
Why It Matters
A 10ms cache hit at the gateway can replace a 500ms backend call. For high-traffic endpoints, this translates to significant cost savings and latency improvements.
Real-World Use
Durga Antivirus Pro caches threat intelligence queries at the gateway for 60 seconds, serving thousands of identical requests from cache while reducing load on the analytics service by 80 percent.
Gateway Caching Flow
flowchart LR
Request-->Gateway
Gateway-->Cache{Cache Hit?}
Cache-->|Yes|Response
Cache-->|No|Backend[Backend Service]
Backend-->Stale[Stale Cache]
Stale-->Response
Backend-->Fresh[Fresh Cache]
Fresh-->Cache
Backend-->Response
Response Cache Implementation
A gateway response cache stores responses keyed by request method and path with optional query parameters.
import time
import hashlib
import json
from typing import Dict, Optional, Any
from collections import OrderedDict
class GatewayResponseCache:
def __init__(self, max_size: int = 1000,
default_ttl: int = 60):
self.cache: OrderedDict[str, Dict] = OrderedDict()
self.max_size = max_size
self.default_ttl = default_ttl
def _build_key(self, method: str, path: str,
headers: Optional[Dict] = None,
query: Optional[Dict] = None) -> str:
components = [method, path]
if query:
sorted_query = sorted(query.items())
components.append(json.dumps(sorted_query))
if headers:
relevant = {
k: v for k, v in headers.items()
if k.lower() in (
"accept", "accept-language",
"authorization"
)
}
if relevant:
components.append(json.dumps(relevant))
raw = "|".join(components)
return hashlib.sha256(raw.encode()).hexdigest()
def get(self, method: str, path: str,
headers: Optional[Dict] = None,
query: Optional[Dict] = None
) -> Optional[Dict]:
key = self._build_key(method, path, headers, query)
entry = self.cache.get(key)
if not entry:
return None
if time.time() > entry["expires"]:
del self.cache[key]
return None
entry["hits"] += 1
return entry["response"]
def set(self, method: str, path: str,
response: Dict,
headers: Optional[Dict] = None,
query: Optional[Dict] = None,
ttl: Optional[int] = None):
key = self._build_key(method, path, headers, query)
if len(self.cache) >= self.max_size:
self.cache.popitem(last=False)
self.cache[key] = {
"response": response,
"expires": time.time() + (ttl or self.default_ttl),
"hits": 0,
"created": time.time()
}
def invalidate(self, path_pattern: str):
keys_to_delete = [
k for k in self.cache if path_pattern in k
]
for k in keys_to_delete:
del self.cache[k]
return len(keys_to_delete)
def stats(self) -> Dict:
total = len(self.cache)
hits = sum(e["hits"] for e in self.cache.values())
return {"entries": total, "total_hits": hits}
cache = GatewayResponseCache(max_size=100)
cache.set("GET", "/api/threats", {"data": "clean"})
result = cache.get("GET", "/api/threats")
print(f"Cached: {result}")
inval_count = cache.invalidate("/api/threats")
print(f"Invalidated: {inval_count} entries")
result = cache.get("GET", "/api/threats")
print(f"After invalidation: {result}")
Cache Key Strategies
Different API patterns require different cache key strategies for correctness.
from typing import Dict, Optional
import hashlib
import json
class CacheKeyBuilder:
@staticmethod
def exact_path(method: str, path: str) -> str:
return f"{method}:{path}"
@staticmethod
def with_query(method: str, path: str,
query: Dict) -> str:
sorted_q = sorted(query.items())
return f"{method}:{path}?{json.dumps(sorted_q)}"
@staticmethod
def with_auth(method: str, path: str,
user_id: str) -> str:
return f"{method}:{path}:user:{user_id}"
@staticmethod
def with_vary(method: str, path: str,
vary_headers: Dict) -> str:
raw = f"{method}:{path}:{json.dumps(vary_headers)}"
return hashlib.md5(raw.encode()).hexdigest()
builder = CacheKeyBuilder()
k1 = builder.exact_path("GET", "/api/health")
k2 = builder.with_query("GET", "/api/search", {"q": "virus"})
k3 = builder.with_auth("GET", "/api/profile", "user-42")
print(f"Key 1: {k1}")
print(f"Key 2: {k2}")
print(f"Key 3: {k3}")
Stale-While-Revalidate
Serve stale content while refreshing the cache in the background to improve perceived latency.
import time
from typing import Dict, Optional, Callable, Any
import threading
class StaleWhileRevalidate:
def __init__(self, backend_call: Callable,
ttl: int = 60, stale_ttl: int = 300):
self.backend_call = backend_call
self.ttl = ttl
self.stale_ttl = stale_ttl
self.cache: Dict[str, Dict] = {}
self.refreshing: set = set()
def get(self, key: str) -> Dict:
now = time.time()
entry = self.cache.get(key)
if entry and now < entry["expires"]:
return entry["response"]
if entry and now < entry["stale_expires"]:
if key not in self.refreshing:
self.refreshing.add(key)
thread = threading.Thread(
target=self._refresh,
args=(key,),
daemon=True
)
thread.start()
return entry["response"]
return self._fetch(key)
def _refresh(self, key: str):
try:
response = self.backend_call(key)
now = time.time()
self.cache[key] = {
"response": response,
"expires": now + self.ttl,
"stale_expires": now + self.stale_ttl
}
finally:
self.refreshing.discard(key)
def _fetch(self, key: str) -> Dict:
response = self.backend_call(key)
now = time.time()
self.cache[key] = {
"response": response,
"expires": now + self.ttl,
"stale_expires": now + self.stale_ttl
}
return response
def fetch_from_backend(key):
print(f"Fetching {key} from backend...")
time.sleep(0.1)
return {"data": f"result for {key}"}
cache = StaleWhileRevalidate(fetch_from_backend, ttl=1, stale_ttl=10)
result1 = cache.get("threats")
print(f"Fresh: {result1}")
time.sleep(1.5)
result2 = cache.get("threats")
print(f"Stale (revalidating): {result2}")
Common Mistakes
Mistake 1: Caching Authenticated Responses Without User Isolation
Caching a user-specific response and serving it to another user leaks data. Always include user context in cache keys.
Mistake 2: Ignoring Cache Invalidation
Without invalidation, clients receive stale data. Implement purge endpoints or use event-driven invalidation.
Mistake 3: Caching Error Responses
Never cache 4xx or 5xx responses. A transient error cached and served for minutes multiplies the impact.
Mistake 4: Overly Long TTLs
Long TTLs improve hit ratio but increase staleness. Match TTL to how frequently the underlying data changes.
Mistake 5: Not Varying on Accept-Encoding
Caching responses without considering content encoding returns gzipped content to clients that do not support it.
Practice Questions
- What is the difference between a cache hit and a cache miss at the gateway?
- Why should cache keys include relevant request headers?
- How does stale-while-revalidate improve perceived latency?
- What is cache poisoning and how does it relate to gateway caching?
- How do you handle cache invalidation across multiple gateway instances?
Challenge
Build a gateway caching layer that supports per-endpoint TTLs, user-aware cache keys, stale-while-revalidate with background refresh, and a purge endpoint for manual invalidation.
FAQ
Mini Project
Build a gateway cache plugin that supports configurable TTL per path pattern, user-aware and anonymous cache keys, Redis backend for distributed caching, stale-while-revalidate, and a Prometheus metrics endpoint for hit ratio monitoring.
What's Next
Learn about CDN Caching for edge caching strategies, or explore Redis Cache for distributed cache backend configuration.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro