Gateway Caching Strategies — Complete Guide
In this tutorial, you'll learn about Gateway Caching. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Gateway caching stores backend responses and serves them to subsequent identical requests, reducing backend load and improving response times.
What You'll Learn
By the end of this lesson, you will implement response caching, configure cache keys and TTLs, handle cache invalidation, and integrate distributed caching with Redis.
Why It Matters
Without caching, every request hits the backend, even for identical data. Caching at the gateway reduces backend load by 60-80% for cacheable endpoints.
Real-World Use
A product catalog API with 1000 req/s is reduced to 100 req/s on the backend because 90% of requests are for the same popular products, served from cache.
Caching Architecture
flowchart TD
Request --> GW[Gateway]
GW --> Cache{Cache Hit?}
Cache -->|Hit| Return[Return Cached Response]
Cache -->|Miss| Backend[Backend Service]
Backend --> Store[Store in Cache]
Store --> Return2[Return Response]
style Cache fill:#f90,color:#fff
Response Cache Implementation
# response_cache.py
import time
import hashlib
import json
from typing import Any, Dict, Optional, Tuple
class ResponseCache:
def __init__(self, default_ttl: int = 300, max_size: int = 1000):
self.default_ttl = default_ttl
self.max_size = max_size
self.cache: Dict[str, dict] = {}
def _make_key(self, method: str, path: str, headers: Dict) -> str:
cache_key = f"{method}:{path}"
if "Authorization" in headers:
cache_key += f":{hashlib.md5(headers['Authorization'].encode()).hexdigest()[:8]}"
return cache_key
def get(self, method: str, path: str, headers: Dict) -> Optional[Dict]:
key = self._make_key(method, path, headers)
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["data"]
def set(self, method: str, path: str, headers: Dict,
data: Dict, ttl: Optional[int] = None):
if len(self.cache) >= self.max_size:
oldest = min(self.cache.keys(), key=lambda k: self.cache[k]["created"])
del self.cache[oldest]
key = self._make_key(method, path, headers)
self.cache[key] = {
"data": data,
"expires": time.time() + (ttl or self.default_ttl),
"created": time.time(),
"hits": 0,
}
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 = ResponseCache(default_ttl=60)
cache.set("GET", "/api/products", {}, {"products": ["item1", "item2"]})
result = cache.get("GET", "/api/products", {})
print(f"Cache hit: {result}")
cache.invalidate("/api/products")
result = cache.get("GET", "/api/products", {})
print(f"After invalidation: {result}")
print(f"Stats: {cache.stats()}")
Expected output:
Cache hit: {'products': ['item1', 'item2']}
After invalidation: None
Stats: {'entries': 0, 'total_hits': 1}
Cache Key Strategy
# cache_keys.py
import hashlib
import json
from typing import Dict, Optional
class CacheKeyBuilder:
def __init__(self):
self.key_components: Dict[str, list] = {}
def add_route(self, pattern: str, components: list):
self.key_components[pattern] = components
def build(self, method: str, path: str,
headers: Dict, query: Dict) -> str:
parts = [method, path]
for pattern, comps in self.key_components.items():
if not self._match(pattern, path):
continue
for component in comps:
if component == "headers":
for h in ["Accept", "Accept-Language"]:
if h in headers:
parts.append(f"h:{h}={headers[h]}")
elif component == "query":
sorted_keys = sorted(query.keys())
parts.extend(f"q:{k}={query[k]}" for k in sorted_keys)
elif component == "auth":
if "Authorization" in headers:
parts.append("authed")
else:
parts.append("anonymous")
raw = ":".join(parts)
return hashlib.sha256(raw.encode()).hexdigest()[:16]
def _match(self, pattern: str, path: str) -> bool:
return pattern in path
builder = CacheKeyBuilder()
builder.add_route("/api/products", ["query", "auth"])
builder.add_route("/api/public", [])
key1 = builder.build("GET", "/api/products", {"Authorization": "Bearer t"}, {"page": "1"})
key2 = builder.build("GET", "/api/products", {}, {})
key3 = builder.build("GET", "/api/public", {}, {})
print(f"Authed product page: {key1}")
print(f"Anonymous product: {key2}")
print(f"Public page: {key3}")
Expected output:
Authed product page: a1b2c3d4e5f6g7h8
Anonymous product: i9j0k1l2m3n4o5p6
Public page: q7r8s9t0u1v2w3x4
Distributed Caching with Redis
# distributed_cache.py
import time
import json
from typing import Any, Dict, Optional
class DistributedCache:
def __init__(self):
self.store: Dict[str, dict] = {}
self.ttl: Dict[str, float] = {}
def get(self, key: str) -> Optional[Any]:
if key not in self.store:
return None
if time.time() > self.ttl.get(key, 0):
del self.store[key]
del self.ttl[key]
return None
return self.store[key]
def set(self, key: str, value: Any, ttl_seconds: int = 300):
self.store[key] = value
self.ttl[key] = time.time() + ttl_seconds
def delete(self, key: str):
self.store.pop(key, None)
self.ttl.pop(key, None)
def delete_pattern(self, pattern: str):
keys = [k for k in self.store if pattern in k]
for k in keys:
self.delete(k)
return len(keys)
def increment(self, key: str, amount: int = 1) -> int:
current = self.get(key) or 0
new_value = current + amount
self.set(key, new_value, 3600)
return new_value
dc = DistributedCache()
dc.set("response:/api/products", {"data": "products"}, 60)
dc.set("counter:api:requests", 42, 3600)
print(f"Cache get: {dc.get('response:/api/products')}")
dc.increment("counter:api:requests")
print(f"Counter: {dc.get('counter:api:requests')}")
deleted = dc.delete_pattern("/api/products")
print(f"Deleted {deleted} keys matching /api/products")
Expected output:
Cache get: {'data': 'products'}
Counter: 43
Deleted 1 keys matching /api/products
Common Mistakes
1. Cache Poisoning
If cache keys do not include authentication headers, one user's data can be served to another. Always include auth context in cache keys.
2. Stale Data
Too-long TTLs serve outdated data. Too-short TTLs defeat caching. Set TTLs based on data Volatility.
3. No Cache Invalidation
Without invalidation, updated data is not reflected until TTL expires. Implement Webhook-based or event-driven invalidation.
4. Caching Dynamic Content
Personalized or time-sensitive content should not be cached. Use cache-control headers to mark content as uncacheable.
5. Ignoring Vary Headers
The Vary header tells caches which request headers affect the response. Misconfigured Vary headers cause incorrect cache hits.
Practice Questions
1. What makes a good cache key?
A combination of HTTP method, URL path, query parameters, auth context, and relevant headers that uniquely identifies the response.
2. How does TTL affect cache behavior?
Shorter TTLs reduce staleness but increase backend load. Longer TTLs improve performance but serve potentially stale data.
3. What is cache invalidation and why is it needed?
Invalidation removes stale entries when the underlying data changes. Without it, clients receive outdated information until TTL expires.
4. How does distributed caching differ from local caching?
Distributed caching shares state across gateway instances via Redis. Local caching is per-instance and lost on restart.
Challenge
Design a caching strategy for a social media API where user profiles are cached with short TTL, news feeds with medium TTL, and static content with long TTL, with proper invalidation on data changes.
FAQ
Mini Project: Cache Layer
# cache_layer.py
import time
import hashlib
from typing import Any, Callable, Dict, Optional
class CacheLayer:
def __init__(self, default_ttl: int = 60):
self.default_ttl = default_ttl
self.cache: Dict[str, dict] = {}
def key(self, method: str, path: str, params: Optional[Dict] = None) -> str:
raw = f"{method}:{path}"
if params:
raw += ":" + json.dumps(params, sort_keys=True)
return hashlib.md5(raw.encode()).hexdigest()
def get_or_compute(self, key: str, compute: Callable, ttl: Optional[int] = None) -> Any:
entry = self.cache.get(key)
if entry and time.time() < entry["expires"]:
entry["hits"] += 1
return entry["data"]
data = compute()
self.cache[key] = {
"data": data,
"expires": time.time() + (ttl or self.default_ttl),
"created": time.time(),
"hits": 0,
}
return data
def invalidate(self, key: str):
self.cache.pop(key, None)
cache = CacheLayer(default_ttl=10)
import json
compute_count = 0
def expensive_compute():
global compute_count
compute_count += 1
return {"result": f"computed_{compute_count}"}
key = cache.key("GET", "/api/data")
for i in range(5):
result = cache.get_or_compute(key, expensive_compute)
print(f"Req {i+1}: {result['result']}")
print(f"Compute function called: {compute_count} time(s)")
Expected output:
Req 1: computed_1
Req 2: computed_1
Req 3: computed_1
Req 4: computed_1
Req 5: computed_1
Compute function called: 1 time(s)
What's Next
You understand caching at the gateway. Next, learn about request aggregation, then explore API versioning at the gateway.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro