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

1"""Cache integration for lexigram-admin. 

2 

3Provides ``AdminCacheService`` — a thin, Result-unwrapping facade over 

4``CacheBackendProtocol`` — and ``CachedPermissionService``, which layers a 

5process-local L1 cache in front of an optional distributed backend. 

6 

7Cross-request caching decorators (``@cache``, ``@remember``, 

8``@invalidate_cache``) live in ``lexigram.cache.service.decorators``. 

9Request-scoped caching lives in ``lexigram.cache.service.request_cache``. 

10""" 

11 

12from __future__ import annotations 

13 

14import asyncio 

15from datetime import UTC, datetime, timedelta 

16import hashlib 

17from typing import Any, Protocol, runtime_checkable 

18 

19from lexigram.contracts.infra.cache import CacheBackendProtocol 

20from lexigram.di.decorators import inject 

21from lexigram.serialization import dumps_str 

22 

23# ============================================================================ 

24# Cache Key Builders 

25# ============================================================================ 

26 

27 

28def default_key_builder(func: Callable[..., Any], *args: Any, **kwargs: Any) -> str: # type: ignore[name-defined] 

29 """Build cache key from function and arguments.""" 

30 # Get function identifier 

31 func_name = f"{func.__module__}.{func.__qualname__}" 

32 

33 # Serialize arguments 

34 key_parts = [func_name] 

35 

36 for arg in args: 

37 if hasattr(arg, "id"): 

38 key_parts.append(f"id:{arg.id}") 

39 elif isinstance(arg, (str, int, float, bool)): 

40 key_parts.append(str(arg)) 

41 else: 

42 key_parts.append(hashlib.sha256(dumps_str(arg).encode()).hexdigest()[:8]) 

43 

44 for k, v in sorted(kwargs.items()): 

45 if isinstance(v, (str, int, float, bool)): 

46 key_parts.append(f"{k}:{v}") 

47 else: 

48 key_parts.append( 

49 f"{k}:{hashlib.sha256(dumps_str(v).encode()).hexdigest()[:8]}", 

50 ) 

51 

52 return ":".join(key_parts) 

53 

54 

55# ============================================================================ 

56# Cache Service 

57# ============================================================================ 

58 

59 

60@inject 

61class AdminCacheService: 

62 """Core caching service for the admin panel. 

63 

64 Provides a clean, unified API for caching using a 

65 contract-based cache backend. 

66 

67 Example: 

68 >>> cache = AdminCacheService(cache_backend=my_backend) 

69 >>> await cache.set("user:1", {"name": "Admin"}) 

70 >>> user = await cache.get("user:1") 

71 """ 

72 

73 def __init__( 

74 self, 

75 cache_backend: CacheBackendProtocol, 

76 ) -> None: 

77 """Initialize the service. 

78 

79 Args: 

80 cache_backend: Cache backend to use for storage 

81 """ 

82 self._backend = cache_backend 

83 

84 async def get(self, key: str) -> Any | None: 

85 """Get value from cache.""" 

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

87 return res.unwrap() if res.is_ok() else None 

88 

89 async def set(self, key: str, value: Any, ttl: int | None = None) -> bool: 

90 """Set value in cache.""" 

91 res = await self._backend.set(key, value, ttl) 

92 return res.is_ok() 

93 

94 async def delete(self, key: str) -> bool: 

95 """Delete key from cache.""" 

96 res = await self._backend.delete(key) 

97 return res.unwrap() if res.is_ok() else False 

98 

99 async def delete_pattern(self, pattern: str) -> int: 

100 """Delete keys matching pattern.""" 

101 res = await self._backend.delete_pattern(pattern) 

102 return res.unwrap() if res.is_ok() else 0 

103 

104 async def get_or_set( 

105 self, 

106 key: str, 

107 factory: Any, 

108 ttl: int | None = None, 

109 ) -> Any: 

110 """Get from cache or compute and store.""" 

111 value = await self.get(key) 

112 if value is not None: 

113 return value 

114 

115 # Compute value 

116 if asyncio.iscoroutinefunction(factory): 

117 value = await factory() 

118 else: 

119 value = factory() 

120 

121 await self.set(key, value, ttl) 

122 return value 

123 

124 

125# ============================================================================ 

126# Cached Permission Service 

127# ============================================================================ 

128 

129 

130@runtime_checkable 

131class UserRoleProviderProtocol(Protocol): 

132 """Protocol for user role providers.""" 

133 

134 async def get_roles(self, user: Any) -> list[str]: ... 

135 

136 async def get_role_permissions(self, role: str) -> set[str]: ... 

137 

138 

139@inject 

140class CachedPermissionService: 

141 """Permission service with caching. 

142 

143 Caches user permissions with configurable TTL. 

144 

145 Example: 

146 >>> service = CachedPermissionService(role_provider, ttl=300) 

147 >>> permissions = await service.get_permissions(user) 

148 """ 

149 

150 def __init__( 

151 self, 

152 role_provider: UserRoleProviderProtocol, 

153 cache_backend: CacheBackendProtocol | None = None, 

154 ttl: int = 300, 

155 ): 

156 self._role_provider = role_provider 

157 self._cache: CacheBackendProtocol | None = cache_backend 

158 self._ttl = ttl 

159 self._local_cache: dict[Any, tuple[set[str], datetime]] = {} 

160 

161 async def get_permissions(self, user: Any) -> set[str]: 

162 """Get user permissions with caching.""" 

163 user_id = getattr(user, "id", id(user)) 

164 cache_key = f"admin:perms:{user_id}" 

165 

166 # Check local cache first (process-level) 

167 if user_id in self._local_cache: 

168 perms, expires = self._local_cache[user_id] 

169 if datetime.now(UTC) < expires: 

170 return perms 

171 del self._local_cache[user_id] 

172 

173 # Check distributed cache 

174 if self._cache is not None: 

175 res = await self._cache.get(cache_key) 

176 if res.is_ok(): 

177 cached = res.unwrap() 

178 if cached is not None: 

179 self._local_cache[user_id] = ( 

180 cached, 

181 datetime.now(UTC) + timedelta(seconds=self._ttl // 2), 

182 ) 

183 return cached 

184 

185 # Compute permissions 

186 permissions: set[str] = set() 

187 roles = await self._role_provider.get_roles(user) 

188 

189 for role in roles: 

190 role_permissions = await self._role_provider.get_role_permissions(role) 

191 permissions.update(role_permissions) 

192 

193 # Cache permissions 

194 if self._cache is not None: 

195 await self._cache.set(cache_key, permissions, self._ttl) 

196 self._local_cache[user_id] = ( 

197 permissions, 

198 datetime.now(UTC) + timedelta(seconds=self._ttl // 2), 

199 ) 

200 

201 return permissions 

202 

203 async def invalidate_user(self, user_id: Any) -> None: 

204 """Invalidate cached permissions for a user.""" 

205 cache_key = f"admin:perms:{user_id}" 

206 if self._cache is not None: 

207 await self._cache.delete(cache_key) 

208 self._local_cache.pop(user_id, None) 

209 

210 async def invalidate_all(self) -> None: 

211 """Invalidate all cached permissions.""" 

212 if self._cache is not None: 

213 await self._cache.delete_pattern("admin:perms:*") 

214 self._local_cache.clear() 

215 

216 def has_permission(self, user: Any, permission: str) -> bool: 

217 """Check if user has permission (sync, uses local cache).""" 

218 user_id = getattr(user, "id", id(user)) 

219 if user_id in self._local_cache: 

220 perms, expires = self._local_cache[user_id] 

221 if datetime.now(UTC) < expires: 

222 return permission in perms 

223 return False 

224 

225 

226__all__ = [ 

227 # Service 

228 "AdminCacheService", 

229 # Backend protocol re-export for ergonomic imports 

230 "CacheBackendProtocol", 

231 # Permission caching 

232 "CachedPermissionService", 

233 "UserRoleProviderProtocol", 

234 # Key builders 

235 "default_key_builder", 

236]