Coverage for apps / ai / services / cache.py: 45%

1"""AI response caching utilities using Django cache framework.""" 

2 

3import hashlib 

4import json 

5import logging 

6from functools import wraps 

7from typing import Callable, Optional 

8 

9from django.core.cache import cache 

10 

11logger = logging.getLogger(__name__) 

12 

13# Cache timeout constants (in seconds) 

14CACHE_TIMEOUT_SHORT = 60 * 30 # 30 minutes - for timer names 

15CACHE_TIMEOUT_MEDIUM = 60 * 60 * 4 # 4 hours - for remix suggestions 

16 

17 

18def _make_cache_key(prefix: str, *args, **kwargs) -> str: 

19 """Generate a deterministic cache key from function arguments. 

20 

21 Args: 

22 prefix: A prefix for the cache key (typically function name). 

23 *args: Positional arguments to include in key. 

24 **kwargs: Keyword arguments to include in key. 

25 

26 Returns: 

27 A cache key string like 'ai:prefix:hash'. 

28 """ 

29 # Create a deterministic representation of the arguments 

30 key_data = { 

31 "args": list(args), 

32 "kwargs": sorted(kwargs.items()), 

33 } 

34 key_json = json.dumps(key_data, sort_keys=True, default=str) 

35 

36 # Hash to keep key length manageable 

37 key_hash = hashlib.sha256(key_json.encode()).hexdigest()[:16] 

38 

39 return f"ai:{prefix}:{key_hash}" 

40 

41 

42def cache_ai_response( 

43 prefix: str, 

44 timeout: int = CACHE_TIMEOUT_MEDIUM, 

45 key_args: Optional[list[int]] = None, 

46 key_kwargs: Optional[list[str]] = None, 

47) -> Callable: 

48 """Decorator to cache AI service responses. 

49 

50 Args: 

51 prefix: Cache key prefix (typically function name). 

52 timeout: Cache timeout in seconds. 

53 key_args: Indices of positional args to include in cache key (default: all). 

54 key_kwargs: Names of kwargs to include in cache key (default: all). 

55 

56 Returns: 

57 Decorated function that caches its results. 

58 

59 Example: 

60 @cache_ai_response('timer_name', timeout=1800) 

61 def generate_timer_name(step_text: str, duration_minutes: int) -> dict: 

62 ... 

63 """ 

64 

65 def decorator(func: Callable) -> Callable: 

66 @wraps(func) 

67 def wrapper(*args, **kwargs): 

68 # Build cache key from selected arguments 

69 if key_args is not None: 

70 cache_args = tuple(args[i] for i in key_args if i < len(args)) 

71 else: 

72 cache_args = args 

73 

74 if key_kwargs is not None: 

75 cache_kwargs = {k: v for k, v in kwargs.items() if k in key_kwargs} 

76 else: 

77 cache_kwargs = kwargs 

78 

79 cache_key = _make_cache_key(prefix, *cache_args, **cache_kwargs) 

80 

81 # Check cache 

82 cached_result = cache.get(cache_key) 

83 if cached_result is not None: 

84 logger.debug(f"Cache hit for {prefix}: {cache_key}") 

85 return cached_result 

86 

87 # Call the actual function 

88 result = func(*args, **kwargs) 

89 

90 # Cache the result 

91 cache.set(cache_key, result, timeout) 

92 logger.debug(f"Cached {prefix} result: {cache_key}") 

93 

94 return result 

95 

96 return wrapper 

97 

98 return decorator 

99 

100 

101def invalidate_ai_cache(prefix: str, *args, **kwargs) -> bool: 

102 """Invalidate a specific AI cache entry. 

103 

104 Args: 

105 prefix: Cache key prefix. 

106 *args: Arguments used in the original cache key. 

107 **kwargs: Keyword arguments used in the original cache key. 

108 

109 Returns: 

110 True if a key was deleted, False otherwise. 

111 """ 

112 cache_key = _make_cache_key(prefix, *args, **kwargs) 

113 return cache.delete(cache_key)