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

1"""Adapter that delegates admin caching to ``lexigram-cache``. 

2 

3When ``lexigram-cache`` is installed, ``AdminCacheServiceAdapter`` wraps 

4``lexigram.cache.service.core.CacheService`` for rich caching features 

5(stampede protection, tag invalidation, batch ops). When not installed, 

6it falls back to a no-op that satisfies the same protocol. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import Any 

12 

13try: 

14 from lexigram.cache.service.core import CacheService as _LexigramCacheService 

15 from lexigram.contracts.infra.cache import CacheBackendProtocol 

16 

17 _HAS_CACHE = True 

18except ImportError: # noqa: F841 

19 _HAS_CACHE = False 

20 _LexigramCacheService = object # type: ignore[assignment,misc] 

21 CacheBackendProtocol = object # type: ignore[assignment,misc] 

22 

23 

24class AdminCacheServiceAdapter: 

25 """Bridge between admin's cache needs and ``lexigram-cache``'s ``CacheService``. 

26 

27 Usage:: 

28 

29 adapter = AdminCacheServiceAdapter(cache_service=real_service) 

30 value = await adapter.get("key") 

31 await adapter.set("key", value, ttl=60) 

32 """ 

33 

34 def __init__( 

35 self, 

36 cache_service: Any | None = None, 

37 ) -> None: 

38 self._service = cache_service 

39 

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

41 """Get a value from cache.""" 

42 if self._service is None: 

43 return None 

44 try: 

45 return await self._service.get(key) 

46 except Exception: # noqa: BLE001 

47 return None 

48 

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

50 """Set a value in cache.""" 

51 if self._service is None: 

52 return False 

53 try: 

54 await self._service.set(key, value, ttl) 

55 return True 

56 except Exception: # noqa: BLE001 

57 return False 

58 

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

60 """Delete a key from cache.""" 

61 if self._service is None: 

62 return False 

63 try: 

64 await self._service.delete(key) 

65 return True 

66 except Exception: # noqa: BLE001 

67 return False 

68 

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

70 """Delete all keys matching a pattern.""" 

71 if self._service is None: 

72 return 0 

73 try: 

74 return await self._service.delete_pattern(pattern) 

75 except Exception: # noqa: BLE001 

76 return 0 

77 

78 async def get_or_set( 

79 self, 

80 key: str, 

81 factory: Any, 

82 ttl: int | None = None, 

83 ) -> Any: 

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

85 if self._service is None: 

86 return await factory() if callable(factory) else factory 

87 try: 

88 return await self._service.get_or_set(key, factory, ttl) 

89 except Exception: # noqa: BLE001 

90 return await factory() if callable(factory) else factory 

91 

92 async def exists(self, key: str) -> bool: 

93 """Check if a key exists in cache.""" 

94 if self._service is None: 

95 return False 

96 try: 

97 return await self._service.exists(key) 

98 except Exception: # noqa: BLE001 

99 return False 

100 

101 async def clear(self) -> bool: 

102 """Clear all cached entries.""" 

103 if self._service is None: 

104 return False 

105 try: 

106 await self._service.clear() 

107 return True 

108 except Exception: # noqa: BLE001 

109 return False 

110 

111 

112__all__ = ["AdminCacheServiceAdapter"]