Coverage for src / lexigram / admin / middleware / cache.py: 0%

1"""Caching middleware for Lexigram Admin.""" 

2 

3from __future__ import annotations 

4 

5import re 

6from typing import Any 

7 

8from starlette.types import ASGIApp, Receive, Scope, Send 

9 

10from lexigram.contracts.infra.cache import CacheBackendProtocol 

11from lexigram.di.decorators import inject 

12from lexigram.logging import get_logger 

13 

14_log = get_logger(__name__) 

15 

16 

17@inject 

18class AdminCacheMiddleware: 

19 """Pure ASGI middleware for response caching backed by ``CacheBackendProtocol``. 

20 

21 10x faster than BaseHTTPMiddleware — no task-creation overhead. 

22 

23 Caches successful GET responses keyed by path, query string, and role. 

24 TTL is determined by the response ``Cache-Control: max-age`` header; the 

25 ``ttl`` constructor argument is used as the default when no header is 

26 present. The backend manages expiration — no manual timestamp tracking. 

27 

28 Args: 

29 app: ASGI application. 

30 cache_backend: Cache backend for response storage. Required for caching. 

31 ttl: Default cache TTL in seconds. 

32 config: Optional ``CacheConfig`` object. 

33 settings_service: Optional ``SettingsService`` for runtime overrides. 

34 """ 

35 

36 def __init__( 

37 self, 

38 app: ASGIApp, 

39 cache_backend: CacheBackendProtocol | None = None, 

40 ttl: int = 60, 

41 config: Any = None, 

42 settings_service: Any = None, 

43 ) -> None: 

44 self.app = app 

45 self.settings_service = settings_service 

46 self._backend: CacheBackendProtocol | None = cache_backend 

47 

48 if config: 

49 self.enabled = getattr(config, "enabled", True) 

50 

51 val = getattr(config, "default_ttl", None) 

52 if val is None and hasattr(config, "get_default_backend"): 

53 backend_cfg = config.get_default_backend() 

54 if backend_cfg: 

55 val = backend_cfg.default_ttl 

56 

57 self.ttl = val if val is not None else ttl 

58 self.excluded_paths = getattr(config, "excluded_paths", []) 

59 else: 

60 self.enabled = True 

61 self.ttl = ttl 

62 self.excluded_paths = [] 

63 

64 if self._backend is None: 

65 self.enabled = False 

66 

67 self._exclusion_patterns = [ 

68 re.compile(p.replace("*", ".*")) for p in self.excluded_paths 

69 ] 

70 

71 def _get_cache_key(self, scope: Scope) -> str: 

72 """Generate cache key from request scope, namespaced by user.""" 

73 path = scope.get("path", "") 

74 query = scope.get("query_string", b"").decode() 

75 state = scope.get("state", {}) or {} 

76 user = state.get("user", None) 

77 identity = "guest" 

78 if user is not None: 

79 user_id = getattr(user, "user_id", None) or getattr(user, "id", None) 

80 if user_id: 

81 identity = f"user:{user_id}" 

82 return f"admin:resp:{path}:{query}:{identity}" 

83 

84 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

85 """Pure ASGI middleware implementation.""" 

86 enabled = self.enabled 

87 ttl = self.ttl 

88 

89 if self.settings_service: 

90 try: 

91 val = await self.settings_service.get( 

92 "admin.cache.enabled", self.enabled 

93 ) 

94 if val is not None: 

95 enabled = bool(val) 

96 val = await self.settings_service.get( 

97 "admin.cache.default_ttl", self.ttl 

98 ) 

99 if val is not None: 

100 ttl = int(val) 

101 except (RuntimeError, ValueError, OSError) as exc: 

102 _log.warning( 

103 "admin.cache_middleware.settings_error", 

104 error=str(exc), 

105 ) 

106 

107 if not enabled: 

108 await self.app(scope, receive, send) 

109 return 

110 if self._backend is None: 

111 await self.app(scope, receive, send) 

112 return 

113 

114 if scope["type"] != "http": 

115 await self.app(scope, receive, send) 

116 return 

117 

118 path = scope.get("path", "") 

119 for pattern in self._exclusion_patterns: 

120 if pattern.match(path): 

121 await self.app(scope, receive, send) 

122 return 

123 

124 if scope.get("method", "") != "GET": 

125 await self.app(scope, receive, send) 

126 return 

127 

128 headers = dict(scope.get("headers", [])) 

129 if headers.get(b"cache-control", b"").decode() == "no-cache": 

130 await self.app(scope, receive, send) 

131 return 

132 

133 key = self._get_cache_key(scope) 

134 

135 # Cache lookup — backend manages TTL expiration. 

136 res = await self._backend.get(key) 

137 if res.is_ok(): 

138 cached = res.unwrap() 

139 if cached is not None: 

140 await send( 

141 { 

142 "type": "http.response.start", 

143 "status": cached["status_code"], 

144 "headers": cached["headers"], 

145 }, 

146 ) 

147 await send({"type": "http.response.body", "body": cached["body"]}) 

148 return 

149 

150 # Capture response from upstream application. 

151 response_started = False 

152 response_status = 200 

153 response_headers: list[Any] = [] 

154 response_body = b"" 

155 

156 async def send_with_caching(message: dict[str, Any]) -> None: 

157 nonlocal response_started, response_status, response_headers, response_body 

158 

159 if message["type"] == "http.response.start": 

160 response_started = True 

161 response_status = message.get("status", 200) 

162 response_headers = list(message.get("headers", [])) 

163 

164 elif message["type"] == "http.response.body": 

165 if not response_started: 

166 return 

167 

168 response_body += message.get("body", b"") 

169 

170 if not message.get("more_body", False): 

171 await self._cache_and_send_response( 

172 send, 

173 response_status, 

174 response_headers, 

175 response_body, 

176 key, 

177 ttl, 

178 ) 

179 else: 

180 await send(message) 

181 

182 await self.app(scope, receive, send_with_caching) # type: ignore[arg-type] 

183 

184 async def _cache_and_send_response( 

185 self, 

186 send: Send, 

187 status: int, 

188 headers: list[Any], 

189 body: bytes, 

190 cache_key: str, 

191 default_ttl: int, 

192 ) -> None: 

193 """Cache a 200 response (respecting Cache-Control) then forward it.""" 

194 if status == 200: 

195 effective_ttl = default_ttl 

196 cache_control = "" 

197 for hkey, hval in headers: 

198 if hkey == b"cache-control": 

199 cache_control = hval.decode() 

200 break 

201 

202 if "max-age=" in cache_control: 

203 try: 

204 for part in (p.strip() for p in cache_control.split(",")): 

205 if part.startswith("max-age="): 

206 effective_ttl = int(part.split("=")[1]) 

207 break 

208 except (ValueError, IndexError): 

209 pass 

210 

211 if "no-store" not in cache_control: 

212 payload = { 

213 "status_code": status, 

214 "headers": headers, 

215 "body": body, 

216 } 

217 store_res = await self._backend.set(cache_key, payload, effective_ttl) # type: ignore[union-attr] 

218 if not store_res.is_ok(): 

219 _log.warning( 

220 "admin.cache_middleware.store_failed", 

221 key=cache_key, 

222 error=str(store_res.unwrap_err()), 

223 ) 

224 

225 await send( 

226 {"type": "http.response.start", "status": status, "headers": headers} 

227 ) 

228 await send({"type": "http.response.body", "body": body})