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

1"""Cache integration — wraps Resource.cacheable with a caching layer.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Awaitable, Callable 

6from dataclasses import dataclass 

7from typing import TYPE_CHECKING, Any 

8 

9if TYPE_CHECKING: 

10 from lexigram.contracts.core.di import ( 

11 ContainerRegistrarProtocol, 

12 ContainerResolverProtocol, 

13 ) 

14 

15 

16@dataclass(frozen=True) 

17class CacheableSpec: 

18 ttl_seconds: int | None = None 

19 key_template: str | None = None 

20 invalidate_on_actions: tuple[str, ...] = () 

21 

22 

23class _NoOpCache: 

24 async def get(self, key: str) -> None: 

25 return None 

26 

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

28 return None 

29 

30 async def delete(self, key: str) -> None: 

31 return None 

32 

33 async def get_or_set( 

34 self, key: str, factory: Callable[[], Awaitable[Any]], ttl: int | None = None 

35 ) -> Any: 

36 return await factory() 

37 

38 

39class CacheIntegration: 

40 """Adapter that decorates data-source calls with a cache layer. 

41 

42 Gracefully no-ops when ``lexigram-cache`` is not installed or when the 

43 integration is disabled via config. 

44 """ 

45 

46 def __init__(self, config: Any) -> None: 

47 self._config = config 

48 self._cache: Any = None 

49 self._enabled = False 

50 

51 def register(self, container: ContainerRegistrarProtocol) -> None: 

52 from lexigram.admin.config import CacheIntegrationConfig 

53 from lexigram.admin.integrations._optional import is_installed 

54 

55 cfg = self._config 

56 if not isinstance(cfg, CacheIntegrationConfig): 

57 cfg = CacheIntegrationConfig() 

58 if not cfg.enabled: 

59 self._cache = _NoOpCache() 

60 return 

61 if not is_installed("lexigram.cache"): 

62 self._cache = _NoOpCache() 

63 return 

64 self._enabled = True 

65 # Resolution deferred to boot() 

66 

67 async def boot(self, container: ContainerResolverProtocol) -> None: 

68 if not self._enabled: 

69 return 

70 try: 

71 from lexigram.contracts.infra.cache import CacheBackendProtocol 

72 

73 self._cache = await container.resolve(CacheBackendProtocol) 

74 except Exception: # noqa: BLE001 

75 self._cache = _NoOpCache() 

76 

77 async def shutdown(self) -> None: 

78 pass 

79 

80 async def health_check(self) -> dict[str, Any]: 

81 return { 

82 "status": "healthy" if not isinstance(self._cache, _NoOpCache) else "noop" 

83 } 

84 

85 def cache_key(self, resource_name: str, *parts: str) -> str: 

86 prefix = getattr(self._config, "key_prefix", "admin") 

87 return f"{prefix}:{resource_name}:" + ":".join(parts) 

88 

89 async def get_or_compute( 

90 self, 

91 key: str, 

92 factory: Callable[[], Awaitable[Any]], 

93 ttl: int | None = None, 

94 ) -> Any: 

95 effective_ttl = ttl or getattr(self._config, "default_ttl_seconds", 60) 

96 return await self._cache.get_or_set(key, factory, effective_ttl) 

97 

98 async def invalidate(self, resource_name: str) -> None: 

99 if hasattr(self._cache, "delete_pattern"): 

100 await self._cache.delete_pattern( 

101 getattr(self._config, "key_prefix", "admin") + f":{resource_name}:*" 

102 ) 

103 

104 

105__all__ = ["CacheIntegration", "CacheableSpec"]