Coverage for src / ai_lls_lib / providers / external.py: 93%

1""" 

2External API provider for production phone verification 

3""" 

4 

5import os 

6 

7import httpx 

8from aws_lambda_powertools import Logger 

9 

10from ..core.models import LineType 

11from .exceptions import ProviderError 

12 

13logger = Logger() 

14 

15 

16class ExternalAPIProvider: 

17 """ 

18 Production provider that calls external verification APIs. 

19 Uses landlineremover.com which returns both line type and DNC status in a single call. 

20 """ 

21 

22 def __init__(self, timeout: float = 10.0): 

23 """ 

24 Initialize external API provider. 

25 

26 Args: 

27 timeout: HTTP request timeout in seconds 

28 """ 

29 self.api_key = os.environ.get("LANDLINE_REMOVER_API_KEY", "") 

30 if not self.api_key: 

31 logger.warning("LANDLINE_REMOVER_API_KEY not set") 

32 

33 self.api_url = "https://app.landlineremover.com/api/check-number" 

34 self.timeout = timeout 

35 self.http_client = httpx.Client(timeout=timeout) 

36 

37 def verify_phone(self, phone: str) -> tuple[LineType, bool]: 

38 """ 

39 Verify phone using landlineremover.com API. 

40 

41 This API returns both line type and DNC status in a single call, 

42 which is more efficient than making two separate API calls. 

43 

44 Args: 

45 phone: E.164 formatted phone number 

46 

47 Returns: 

48 Tuple of (line_type, is_on_dnc_list) 

49 

50 Raises: 

51 httpx.HTTPError: For API communication errors 

52 ValueError: For invalid responses 

53 """ 

54 logger.debug(f"Verifying phone {phone[:6]}*** via external API") 

55 

56 if not self.api_key: 

57 raise ValueError("API key not configured") 

58 

59 try: 

60 # Make single API call that returns both line type and DNC status 

61 # The API may redirect, so we need to follow redirects 

62 response = self.http_client.get( 

63 self.api_url, 

64 params={"apikey": self.api_key, "number": phone}, 

65 follow_redirects=True, 

66 ) 

67 

68 # Raise for HTTP errors 

69 response.raise_for_status() 

70 

71 # Parse response 

72 json_response = response.json() 

73 

74 # Extract data from response wrapper 

75 if "data" in json_response: 

76 data = json_response["data"] 

77 else: 

78 data = json_response 

79 

80 # Map line type from API response 

81 line_type = self._map_line_type(data) 

82 

83 # Map DNC status - API uses "DNCType" field 

84 # Values can be "dnc", "clean", etc. 

85 dnc_type = data.get("DNCType", data.get("dnc_type", "")).lower() 

86 is_dnc = dnc_type != "clean" and dnc_type != "" 

87 

88 logger.debug( 

89 f"Verification complete for {phone[:6]}***", 

90 extra={"line_type": line_type.value, "is_dnc": is_dnc, "dnc_type": dnc_type}, 

91 ) 

92 

93 return line_type, is_dnc 

94 

95 except httpx.HTTPStatusError as e: 

96 logger.error(f"API error: {e.response.status_code} - {e.response.text}") 

97 raise ProviderError( 

98 f"API request failed with status {e.response.status_code}", 

99 status_code=e.response.status_code, 

100 ) from e 

101 except httpx.RequestError as e: 

102 logger.error(f"Network error: {str(e)}") 

103 raise ProviderError(f"Network error during API call: {str(e)}") from e 

104 except Exception as e: 

105 logger.error(f"Unexpected error during verification: {str(e)}") 

106 raise 

107 

108 def _map_line_type(self, data: dict) -> LineType: 

109 """ 

110 Map API response to LineType enum. 

111 

112 Args: 

113 data: API response dictionary 

114 

115 Returns: 

116 LineType enum value 

117 """ 

118 # API uses "LineType" (capitalized) field 

119 line_type_str = data.get("LineType", data.get("line_type", "")).lower() 

120 

121 # Map common line types 

122 line_type_map = { 

123 "mobile": LineType.MOBILE, 

124 "landline": LineType.LANDLINE, 

125 "voip": LineType.VOIP, 

126 "wireless": LineType.MOBILE, # Some APIs return "wireless" for mobile 

127 "fixed": LineType.LANDLINE, # Some APIs return "fixed" for landline 

128 } 

129 

130 return line_type_map.get(line_type_str, LineType.UNKNOWN) 

131 

132 def __del__(self) -> None: 

133 """Cleanup HTTP client""" 

134 if hasattr(self, "http_client"): 

135 self.http_client.close()