Coverage for src/lexigram/graphql/core/caching.py: 96%
128 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""GraphQL response caching.
3This module provides response caching to improve performance
4for repeated queries.
5"""
7from __future__ import annotations
9from dataclasses import dataclass
10from datetime import UTC, datetime, timedelta
11from typing import TYPE_CHECKING, Any
13from lexigram import hashing # type: ignore[attr-defined]
14from lexigram import serialization as json
15from lexigram.logging import get_logger
16from lexigram.primitives import clock as ambient_clock
18if TYPE_CHECKING:
19 from lexigram.contracts.infra.cache.protocols import CacheBackendProtocol
22logger = get_logger(__name__)
25class _MemoryCacheShim:
26 """Local in-memory fallback used when no CacheBackendProtocol is registered in the container.
28 Implements the full :class:`~lexigram.contracts.cache.CacheBackendProtocol` protocol so
29 that static analysis and runtime ``isinstance`` checks pass.
31 .. warning::
32 Not suitable for production with multiple process instances — all state is
33 in-process memory with no replication or persistence.
34 """
36 def __init__(self, default_ttl: int = 300) -> None:
37 self._store: dict[str, tuple[Any, datetime | None]] = {}
38 self._default_ttl = timedelta(seconds=default_ttl)
39 self._locks: dict[str, bool] = {}
41 def _now(self) -> datetime:
42 """Get current timestamp."""
43 return ambient_clock.now()
45 async def get(self, key: str) -> Any | None:
46 """Get a value from cache."""
47 if key not in self._store:
48 return None
49 value, expires = self._store[key]
50 if expires and expires < self._now():
51 del self._store[key]
52 return None
53 return value
55 async def set(self, key: str, value: Any, ttl: int | None = None) -> bool:
56 """Set a value in cache."""
57 ttl_delta = timedelta(seconds=ttl) if ttl else self._default_ttl
58 expires = self._now() + ttl_delta if ttl else None
59 self._store[key] = (value, expires)
60 return True
62 async def delete(self, key: str) -> bool:
63 """Delete a value from cache."""
64 existed = key in self._store
65 self._store.pop(key, None)
66 return existed
68 async def delete_many(self, keys: list[str]) -> bool:
69 """Delete multiple values from cache."""
70 for key in keys:
71 self._store.pop(key, None)
72 return True
74 async def exists(self, key: str) -> bool:
75 """Check whether a key exists in cache."""
76 if key not in self._store:
77 return False
78 _, expires = self._store[key]
79 if expires and expires < self._now():
80 del self._store[key]
81 return False
82 return True
84 async def clear(self) -> bool:
85 """Clear all cache entries."""
86 self._store.clear()
87 return True
89 async def get_many(self, keys: list[str]) -> dict[str, Any]:
90 """Get multiple values from cache."""
91 result: dict[str, Any] = {}
92 for key in keys:
93 value = await self.get(key)
94 if value is not None:
95 result[key] = value
96 return result
98 async def set_many(self, items: dict[str, Any], ttl: int | None = None) -> bool:
99 """Set multiple values in cache."""
100 for key, value in items.items():
101 await self.set(key, value, ttl)
102 return True
104 async def acquire_lock(self, key: str, ttl: int) -> bool:
105 """Acquire a simple in-memory lock."""
106 if self._locks.get(key):
107 return False
108 self._locks[key] = True
109 return True
111 async def release_lock(self, key: str) -> bool:
112 """Release an in-memory lock."""
113 existed = self._locks.pop(key, False)
114 return bool(existed)
117@dataclass
118class ResponseCacheEntry:
119 """Cached GraphQL response entry.
121 Attributes:
122 data: The cached response data.
123 extensions: Optional response extensions.
124 created_at: When the entry was created.
125 ttl: Time-to-live in seconds.
126 """
128 data: Any
129 extensions: dict | None = None
130 created_at: datetime | None = None
131 ttl: int | None = None
134def compute_cache_key(
135 query: str,
136 variables: dict | None = None,
137 operation_name: str | None = None,
138 user_id: str | None = None,
139 tenant_id: str | None = None,
140 query_hash: str | None = None,
141) -> str:
142 """Compute a deterministic cache key for a GraphQL operation.
144 The key incorporates the query, variables, operation name, and optionally
145 the user and tenant identifiers so that multi-tenant or user-scoped caches
146 cannot cross-contaminate each other's responses.
148 When APQ (Automatic Persisted Queries) is enabled, callers should pass the
149 pre-computed ``query_hash`` from the APQ extension (SHA-256 hex digest of
150 the query string). This avoids hashing the query twice — once in the APQ
151 pipeline and again here.
153 Args:
154 query: GraphQL query string.
155 variables: Query variables.
156 operation_name: Optional operation name.
157 user_id: Optional user identifier for per-user cache isolation.
158 tenant_id: Optional tenant identifier for multi-tenant isolation.
159 query_hash: Optional pre-computed SHA-256 hex digest of *query* (e.g.
160 from APQ ``extensions.persistedQuery.sha256Hash``). When provided,
161 re-hashing the full query string is skipped.
163 Returns:
164 Namespaced SHA-256 cache key string.
165 """
166 # Use the pre-computed hash when available (APQ fast path) to avoid
167 # re-hashing the query string.
168 base_hash = query_hash or hashing.hash_hex(query)
170 key_parts = [base_hash]
172 if variables:
173 key_parts.append(json.dumps(variables, sort_keys=True).decode("utf-8"))
175 if operation_name:
176 key_parts.append(operation_name)
178 if tenant_id:
179 key_parts.append(f"t:{tenant_id}")
181 if user_id:
182 key_parts.append(f"u:{user_id}")
184 # Only need a second hash pass when there are extra discriminators beyond
185 # the query hash — otherwise base_hash alone is already unique.
186 if len(key_parts) == 1:
187 return f"gql:{base_hash[:32]}"
189 key_string = "|".join(key_parts)
190 return f"gql:{hashing.hash_hex(key_string)[:32]}"
193class ResponseCache:
194 """GraphQL response cache.
196 Caches complete query responses to improve performance for repeated queries.
198 This is a **response-level** cache — the entire query result is cached
199 keyed on ``(query, variables, operation_name)``. This is the current
200 implementation.
202 .. TODO::
203 **Field-level caching via ``@CacheControl`` directive** is not yet
204 implemented. The idea is to annotate individual Strawberry fields with
205 a ``@CacheControl(max_age=60)`` directive so that only the hot, slow
206 resolvers are cached rather than the entire response. This requires:
208 1. A custom Strawberry extension that intercepts field resolution
209 2. Generating ``Cache-Control`` HTTP headers from the directive values
210 (per the Apollo ``@cacheControl`` spec)
211 3. Integrating with :class:`CacheBackendProtocol` for per-field storage
213 Until this is implemented, use ``@functools.lru_cache`` or a
214 ``CacheBackendProtocol`` call inside individual resolver methods as a workaround.
216 Example:
217 ```python
218 from lexigram.contracts.infra.cache.protocols import CacheBackendProtocol
220 cache = ResponseCache(backend=my_cache_backend)
222 # Check cache before execution
223 key = compute_cache_key(query, variables)
224 cached = await cache.get(key)
226 if cached:
227 return cached
229 # Execute query and cache result
230 result = await executor.execute(...)
231 await cache.set(key, result)
232 ```
233 """
235 def __init__(
236 self,
237 backend: CacheBackendProtocol,
238 enabled: bool = True,
239 default_ttl: int = 300,
240 ):
241 """Initialize the cache.
243 Args:
244 backend: Cache backend implementation.
245 enabled: Whether caching is enabled.
246 default_ttl: Default time-to-live in seconds.
247 """
248 self._backend = backend
249 self._enabled = enabled
250 self._default_ttl = default_ttl
252 @property
253 def enabled(self) -> bool:
254 """Check if caching is enabled."""
255 return self._enabled
257 async def get(self, key: str) -> ResponseCacheEntry | None:
258 """Get a cached response.
260 Args:
261 key: Cache key.
263 Returns:
264 Cached entry if found, None otherwise.
265 """
266 if not self._enabled:
267 return None
269 try:
270 return await self._backend.get(key) # type: ignore[return-value]
271 except (OSError, LookupError, RuntimeError) as e:
272 logger.warning("Cache get failed: %s", e)
273 return None
275 async def set(
276 self,
277 key: str,
278 data: Any,
279 extensions: dict | None = None,
280 ttl: int | None = None,
281 ) -> None:
282 """Cache a response.
284 Args:
285 key: Cache key.
286 data: Response data to cache.
287 extensions: Optional response extensions.
288 ttl: Optional time-to-live in seconds.
289 """
290 if not self._enabled:
291 return
293 try:
294 entry = ResponseCacheEntry(
295 data=data,
296 extensions=extensions,
297 created_at=datetime.now(UTC),
298 ttl=ttl or self._default_ttl,
299 )
300 await self._backend.set(key, entry, ttl)
301 except (OSError, LookupError, RuntimeError) as e:
302 logger.warning("Cache set failed: %s", e)
304 async def invalidate(self, key: str) -> None:
305 """Invalidate a cached entry.
307 Args:
308 key: Cache key to invalidate.
309 """
310 try:
311 await self._backend.delete(key)
312 except (OSError, LookupError, RuntimeError) as e:
313 logger.warning("Cache invalidation failed: %s", e)
315 async def clear(self) -> None:
316 """Clear all cached entries."""
317 try:
318 await self._backend.clear()
319 except (OSError, LookupError, RuntimeError) as e:
320 logger.warning("Cache clear failed: %s", e)
323def create_response_cache(
324 backend_type: str = "memory",
325 **kwargs: Any,
326) -> ResponseCache:
327 """Create a response cache with the specified backend.
329 Args:
330 backend_type: Type of backend ('memory').
331 **kwargs: Additional arguments for the backend.
333 Returns:
334 Configured ResponseCache.
335 """
336 if backend_type == "memory":
337 backend: CacheBackendProtocol = _MemoryCacheShim(**kwargs) # type: ignore[assignment]
338 else:
339 raise ValueError(f"Unknown cache backend type: {backend_type}")
341 return ResponseCache(backend=backend)
344__all__ = [
345 "ResponseCache",
346 "ResponseCacheEntry",
347 "compute_cache_key",
348 "create_response_cache",
349]