Caching in GraphQL APIs — Complete Guide
In this tutorial, you'll learn about Caching in Graphql. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
GraphQL caching differs from REST because all requests hit a single POST endpoint, making URL-based HTTP caching ineffective. Solutions include persisted queries, resolver caching, and DataLoader.
What You'll Learn
By the end of this lesson, you will implement persisted queries, use DataLoader for batching, apply resolver-level caching, and configure Apollo Client cache.
Why It Matters
GraphQL's single endpoint prevents traditional HTTP caching. Without proper caching Strategy, every query hits resolvers, increasing database load and latency.
Real-World Use
Apollo Client maintains a normalized in-memory cache. When you query a user by ID, Apollo caches the User type by its id field. Subsequent queries for that user skip the network.
GraphQL Caching Flow
flowchart LR
Query[Query] --> Persisted{Is Persisted?}
Persisted -->|Yes| CDN[CDN Cache]
Persisted -->|No| Parse[Parse Query]
Parse --> DataLoader[DataLoader Batch]
DataLoader --> ResolverCache{Resolver Cache Hit?}
ResolverCache -->|Yes| Return
ResolverCache -->|No| DB[(Database)]
DB --> ResolverCache
ResolverCache --> Return[Response]
Persisted Queries
# persisted_queries.py
import hashlib
from typing import Dict, Optional
class PersistedQueryStore:
def __init__(self):
self.queries: Dict[str, str] = {}
def register(self, query: str) -> str:
q_hash = hashlib.sha256(query.encode()).hexdigest()[:16]
self.queries[q_hash] = query
return q_hash
def lookup(self, q_hash: str) -> Optional[str]:
return self.queries.get(q_hash)
def autotomatic_persist(self, query: str, hash_key: str) -> str:
self.queries[hash_key] = query
return f"Persisted: {hash_key}"
store = PersistedQueryStore()
query = "query { user(id: 1) { name email } }"
q_hash = store.register(query)
print(f"Query hash: {q_hash}")
print(f"Lookup: {store.lookup(q_hash)[:40]}...")
Expected output:
Query hash: <hexstring>
Lookup: query { user(id: 1) { name email } }...
DataLoader for Batching
# dataloader_cache.py
from typing import Any, Callable, Dict, List
class DataLoader:
def __init__(self, batch_fn: Callable):
self.batch_fn = batch_fn
self.queue: List[Any] = []
self.cache: Dict[Any, Any] = {}
def load(self, key: Any) -> Any:
if key in self.cache:
return self.cache[key]
self.queue.append(key)
return None
def drain(self):
if not self.queue:
return
keys = list(set(self.queue))
self.queue = []
results = self.batch_fn(keys)
for key, value in zip(keys, results):
self.cache[key] = value
class UserService:
def __init__(self):
self.db = {
1: {"id": 1, "name": "Alice"},
2: {"id": 2, "name": "Bob"},
3: {"id": 3, "name": "Charlie"},
}
self.query_count = 0
def batch_users(self, ids: List[int]) -> List[Dict]:
self.query_count += 1
return [self.db.get(i, {"id": i, "name": "Unknown"}) for i in ids]
def fetch_users(self, ids: List[int]) -> Dict:
loader = DataLoader(self.batch_users)
results = {}
for uid in ids:
results[uid] = loader.load(uid)
loader.drain()
for uid in ids:
results[uid] = loader.cache[uid]
return results
svc = UserService()
result = svc.fetch_users([1, 2, 1, 3])
print(f"Queries executed: {svc.query_count} (instead of {len([1, 2, 1, 3])})")
for uid, data in result.items():
print(f" User {uid}: {data['name']}")
Expected output:
Queries executed: 1 (instead of 4)
User 1: Alice
User 2: Bob
User 3: Charlie
Apollo Normalized Cache
# normalized_cache.py
from typing import Any, Dict, List, Optional
class NormalizedCache:
def __init__(self):
self.entities: Dict[str, Dict] = {}
self.queries: Dict[str, Any] = {}
def write_fragment(self, type_name: str, id: Any, data: Dict):
key = f"{type_name}:{id}"
if key not in self.entities:
self.entities[key] = {}
self.entities[key].update(data)
def read_fragment(self, type_name: str, id: Any) -> Optional[Dict]:
key = f"{type_name}:{id}"
return self.entities.get(key)
def write_query(self, query_str: str, variables: Dict, result: Any):
q_key = f"{query_str}:{str(variables)}"
self.queries[q_key] = result
def cache_or_fetch(self, query_str: str, variables: Dict,
fetch_fn: callable) -> Any:
q_key = f"{query_str}:{str(variables)}"
if q_key in self.queries:
return {"from_cache": True, "data": self.queries[q_key]}
data = fetch_fn(variables)
self.write_query(query_str, variables, data)
return {"from_cache": False, "data": data}
cache = NormalizedCache()
cache.write_fragment("User", 1, {"id": 1, "name": "Alice", "email": "a@x.com"})
user = cache.read_fragment("User", 1)
print(f"Cached user: {user}")
user2 = cache.read_fragment("User", 99)
print(f"Non-existent user: {user2}")
Expected output:
Cached user: {'id': 1, 'name': 'Alice', 'email': 'a@x.com'}
Non-existent user: None
Common Mistakes
1. Expecting HTTP Caching to Work
GraphQL uses POST for queries. POST responses are not cached by browsers or CDNs without special configuration.
2. No DataLoader
Without DataLoader, resolving a list of N items triggers N database queries. DataLoader batches them.
3. Ignoring Normalized Cache
Storing full query responses duplicates data. Normalize by entity type and ID for efficient updates.
4. Cache Poisoning with Variables
Query results differ by variable values. Include variables in cache keys to avoid serving wrong data.
5. Over-Caching User-Specific Data
Caching user-specific GraphQL responses can leak data. Use query-specific cache keys that include user ID.
Practice Questions
1. Why is HTTP caching harder in GraphQL?
All queries hit the same POST endpoint. URL-based caching cannot distinguish between different queries.
2. What does persisted queries solve?
Clients send a hash instead of the full query string. CDNs can cache responses by hash.
3. What is DataLoader used for?
Batching and caching database queries within a single GraphQL request to avoid N+1 queries.
4. How does Apollo Client cache work?
Normalized cache stores entities by type and ID. When one query updates a user, all queries showing that user update automatically.
Challenge
Build a GraphQL caching layer with persisted queries, DataLoader batching, and normalized entity cache that handles cache invalidation when data is mutated.
FAQ
Mini Project: GraphQL Cache Layer
# graphql_cache.py
from typing import Any, Callable, Dict, List
class GQLCache:
def __init__(self):
self.cache: Dict[str, Any] = {}
self.pending: Dict[str, List[Callable]] = {}
def get_or_fetch(self, key: str, fetcher: Callable) -> Any:
if key in self.cache:
return {"source": "cache", "data": self.cache[key]}
data = fetcher()
self.cache[key] = data
return {"source": "fetched", "data": data}
def invalidate(self, key: str):
self.cache.pop(key, None)
cache = GQLCache()
def fetch_users():
return [{"id": 1, "name": "Alice"}]
r1 = cache.get_or_fetch("users", fetch_users)
r2 = cache.get_or_fetch("users", fetch_users)
print(f"First: {r1['source']}")
print(f"Second: {r2['source']}")
cache.invalidate("users")
r3 = cache.get_or_fetch("users", fetch_users)
print(f"After invalidate: {r3['source']}")
Expected output:
First: fetched
Second: cache
After invalidate: fetched
What's Next
You understand GraphQL caching. Next, explore REST versioning, then GraphQL versioning.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro