Coverage for src/ai_lls_lib/core/cache.py: 100%

1""" 

2DynamoDB cache implementation for phone verifications 

3""" 

4 

5import threading 

6from datetime import UTC, datetime, timedelta 

7from typing import Any 

8 

9import boto3 

10from aws_lambda_powertools import Logger 

11 

12from .models import LineType, PhoneVerification, VerificationSource 

13 

14logger = Logger() 

15 

16 

17class DynamoDBCache: 

18 """Cache for phone verification results using DynamoDB with TTL""" 

19 

20 def __init__(self, table_name: str, ttl_days: int = 90): 

21 self.table_name = table_name 

22 self.ttl_days = ttl_days 

23 # boto3 resources are not thread-safe. The bulk processor verifies rows from a 

24 # worker pool, so each thread gets its own resource and Table, created lazily. 

25 self._local = threading.local() 

26 

27 @property 

28 def dynamodb(self) -> Any: 

29 if not hasattr(self._local, "dynamodb"): 

30 self._local.dynamodb = boto3.resource("dynamodb") 

31 return self._local.dynamodb 

32 

33 @property 

34 def table(self) -> Any: 

35 if not hasattr(self._local, "table"): 

36 self._local.table = self.dynamodb.Table(self.table_name) 

37 return self._local.table 

38 

39 def get(self, phone_number: str) -> PhoneVerification | None: 

40 """Get cached verification result""" 

41 try: 

42 response = self.table.get_item(Key={"phone_number": phone_number}) 

43 

44 if "Item" not in response: 

45 logger.info(f"Cache miss for {phone_number[:6]}***") 

46 return None 

47 

48 item: dict[str, Any] = response["Item"] 

49 

50 if "known_litigator" not in item: 

51 # Entry predates litigator tracking - treat as a miss so the 

52 # number is re-verified and the entry rewritten with the new schema 

53 logger.info( 

54 f"Cache entry for {phone_number[:6]}*** predates known_litigator; " 

55 "treating as miss" 

56 ) 

57 return None 

58 

59 logger.info(f"Cache hit for {phone_number[:6]}***") 

60 

61 return PhoneVerification( 

62 phone_number=str(item["phone_number"]), 

63 line_type=LineType(str(item["line_type"])), 

64 dnc=bool(item["dnc"]), 

65 known_litigator=bool(item["known_litigator"]), 

66 cached=True, 

67 verified_at=datetime.fromisoformat(str(item["verified_at"])), 

68 source=VerificationSource.CACHE, 

69 ) 

70 

71 except Exception as e: 

72 logger.error(f"Cache get error: {str(e)}") 

73 return None 

74 

75 def set(self, phone_number: str, verification: PhoneVerification) -> None: 

76 """Store verification result in cache""" 

77 try: 

78 ttl = int((datetime.now(UTC) + timedelta(days=self.ttl_days)).timestamp()) 

79 

80 self.table.put_item( 

81 Item={ 

82 "phone_number": phone_number, 

83 "line_type": verification.line_type.value, 

84 "dnc": verification.dnc, 

85 "known_litigator": verification.known_litigator, 

86 "verified_at": verification.verified_at.isoformat(), 

87 "source": verification.source.value, 

88 "ttl": ttl, 

89 } 

90 ) 

91 

92 logger.info(f"Cached result for {phone_number[:6]}***") 

93 

94 except Exception as e: 

95 logger.error(f"Cache set error: {str(e)}") 

96 # Don't fail the request if cache write fails 

97 

98 def batch_get(self, phone_numbers: list[str]) -> dict[str, PhoneVerification | None]: 

99 """Get multiple cached results""" 

100 results: dict[str, PhoneVerification | None] = {} 

101 

102 # DynamoDB batch get (max 100 items per request) 

103 for i in range(0, len(phone_numbers), 100): 

104 batch = phone_numbers[i : i + 100] 

105 

106 try: 

107 response = self.dynamodb.batch_get_item( 

108 RequestItems={ 

109 self.table_name: {"Keys": [{"phone_number": phone} for phone in batch]} 

110 } 

111 ) 

112 

113 legacy_entries = 0 

114 for item in response.get("Responses", {}).get(self.table_name, []): 

115 if "known_litigator" not in item: 

116 # Entry predates litigator tracking - leave as a miss 

117 legacy_entries += 1 

118 continue 

119 phone = str(item["phone_number"]) 

120 results[phone] = PhoneVerification( 

121 phone_number=phone, 

122 line_type=LineType(str(item["line_type"])), 

123 dnc=bool(item["dnc"]), 

124 known_litigator=bool(item["known_litigator"]), 

125 cached=True, 

126 verified_at=datetime.fromisoformat(str(item["verified_at"])), 

127 source=VerificationSource.CACHE, 

128 ) 

129 

130 if legacy_entries: 

131 logger.info( 

132 f"{legacy_entries} cache entries predate known_litigator; " 

133 "treating as misses" 

134 ) 

135 

136 except Exception as e: 

137 logger.error(f"Batch cache get error: {str(e)}") 

138 

139 # Fill in None for misses 

140 for phone in phone_numbers: 

141 if phone not in results: 

142 results[phone] = None 

143 

144 return results