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

1""" 

2DynamoDB cache implementation for phone verifications 

3""" 

4 

5from datetime import UTC, datetime, timedelta 

6from typing import Any 

7 

8import boto3 

9from aws_lambda_powertools import Logger 

10 

11from .models import LineType, PhoneVerification, VerificationSource 

12 

13logger = Logger() 

14 

15 

16class DynamoDBCache: 

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

18 

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

20 self.table_name = table_name 

21 self.ttl_days = ttl_days 

22 self.dynamodb = boto3.resource("dynamodb") 

23 self.table = self.dynamodb.Table(table_name) 

24 

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

26 """Get cached verification result""" 

27 try: 

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

29 

30 if "Item" not in response: 

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

32 return None 

33 

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

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

36 

37 return PhoneVerification( 

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

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

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

41 cached=True, 

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

43 source=VerificationSource.CACHE, 

44 ) 

45 

46 except Exception as e: 

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

48 return None 

49 

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

51 """Store verification result in cache""" 

52 try: 

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

54 

55 self.table.put_item( 

56 Item={ 

57 "phone_number": phone_number, 

58 "line_type": verification.line_type.value, 

59 "dnc": verification.dnc, 

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

61 "source": verification.source.value, 

62 "ttl": ttl, 

63 } 

64 ) 

65 

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

67 

68 except Exception as e: 

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

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

71 

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

73 """Get multiple cached results""" 

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

75 

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

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

78 batch = phone_numbers[i : i + 100] 

79 

80 try: 

81 response = self.dynamodb.batch_get_item( 

82 RequestItems={ 

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

84 } 

85 ) 

86 

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

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

89 results[phone] = PhoneVerification( 

90 phone_number=phone, 

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

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

93 cached=True, 

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

95 source=VerificationSource.CACHE, 

96 ) 

97 

98 except Exception as e: 

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

100 

101 # Fill in None for misses 

102 for phone in phone_numbers: 

103 if phone not in results: 

104 results[phone] = None 

105 

106 return results