Coverage for src/lexigram/web/interceptors/builtin/cache.py: 24%

1"""Cache interceptor for HTTP response caching. 

2 

3Provides HTTP response caching based on cache headers. 

4""" 

5 

6from __future__ import annotations 

7 

8import hashlib 

9from typing import TYPE_CHECKING, Any 

10 

11from lexigram.primitives import clock as ambient_clock 

12from lexigram.web.protocols import ( 

13 CallHandlerProtocol, 

14 ExecutionContextProtocol, 

15 WebInterceptorProtocol, 

16) 

17 

18if TYPE_CHECKING: 

19 from collections.abc import Callable 

20 

21 

22class CacheInterceptor(WebInterceptorProtocol): 

23 """Interceptor that handles HTTP response caching. 

24 

25 Respects Cache-Control headers and can cache responses based on 

26 request path and other parameters. 

27 

28 Example: 

29 ```python 

30 # Global cache with 60 second TTL 

31 router.add_interceptor(CacheInterceptor(ttl=60)) 

32 ``` 

33 """ 

34 

35 def __init__( 

36 self, 

37 ttl: int = 300, 

38 cache_key_builder: Callable[[Any], str] | None = None, 

39 ): 

40 """Initialize the cache interceptor. 

41 

42 Args: 

43 ttl: Default time-to-live in seconds for cached responses. 

44 cache_key_builder: Optional custom function to build cache keys. 

45 """ 

46 self._ttl = ttl 

47 self._cache_key_builder = cache_key_builder or self._default_cache_key 

48 self._cache: dict[str, tuple[Any, float]] = {} 

49 

50 async def intercept( 

51 self, 

52 context: ExecutionContextProtocol, 

53 next_handler: CallHandlerProtocol, 

54 ) -> Any: 

55 """Intercept and handle caching. 

56 

57 Args: 

58 context: The execution context. 

59 next_handler: The next handler in the chain. 

60 

61 Returns: 

62 Cached response or fresh response from handler. 

63 """ 

64 request = context.request 

65 

66 # Check if this is a cacheable method 

67 method = getattr(request, "method", "GET") 

68 if method != "GET": 

69 return await next_handler.handle() 

70 

71 # Build cache key 

72 cache_key = self._cache_key_builder(request) 

73 

74 # Check cache 

75 if cache_key in self._cache: 

76 result, expiry = self._cache[cache_key] 

77 timestamp = ambient_clock.timestamp() 

78 if timestamp < expiry: 

79 # Add cache hit header 

80 if hasattr(result, "headers"): 

81 result.headers["X-Cache"] = "HIT" 

82 return result 

83 # Expired - remove from cache 

84 del self._cache[cache_key] 

85 

86 # Execute handler 

87 result = await next_handler.handle() 

88 

89 # Check if response is cacheable 

90 if result is not None and self._is_cacheable(result): 

91 timestamp = ambient_clock.timestamp() 

92 # Store in cache 

93 self._cache[cache_key] = (result, timestamp + self._ttl) 

94 

95 # Add cache miss header 

96 if hasattr(result, "headers"): 

97 result.headers["X-Cache"] = "MISS" 

98 

99 return result 

100 

101 def _default_cache_key(self, request: Any) -> str: 

102 """Build default cache key from request. 

103 

104 Args: 

105 request: The HTTP request. 

106 

107 Returns: 

108 Cache key string. 

109 """ 

110 url = getattr(request, "url", None) 

111 path = url.path if url is not None else "/" 

112 

113 # Include query params in cache key 

114 query_params = str(getattr(request, "query_params", "")) 

115 

116 key_data = f"{path}:{query_params}" 

117 return hashlib.md5(key_data.encode(), usedforsecurity=False).hexdigest() 

118 

119 def _is_cacheable(self, result: Any) -> bool: 

120 """Check if a response is cacheable. 

121 

122 Args: 

123 result: The response to check. 

124 

125 Returns: 

126 True if the response is cacheable. 

127 """ 

128 # Check status code 

129 status_code = getattr(result, "status_code", 200) 

130 if status_code != 200: 

131 return False 

132 

133 # Check Cache-Control header 

134 if hasattr(result, "headers"): 

135 cache_control = result.headers.get("cache-control", "") 

136 if "no-store" in cache_control.lower(): 

137 return False 

138 if "private" in cache_control.lower(): 

139 return False 

140 

141 return True 

142 

143 def clear_cache(self) -> None: 

144 """Clear all cached responses.""" 

145 self._cache.clear() 

146 

147 

148__all__ = ["CacheInterceptor"]