Coverage for src/ai_lls_lib/providers/external.py: 94%
180 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-02 17:06 +0000
1"""
2External API provider for production phone verification
3"""
5import os
6import random
7import threading
8import time
9from collections.abc import Callable
11import httpx
12from aws_lambda_powertools import Logger
14from ..core.models import LineType, VerificationResult
15from .exceptions import (
16 ProviderAuthError,
17 ProviderError,
18 ProviderQuotaError,
19 ProviderUnavailableError,
20)
22logger = Logger()
24RETRYABLE_STATUSES = frozenset({408, 429, 500, 502, 503, 504})
25RETRY_AFTER_CAP_SECONDS = 60.0
27# Best-effort classification of the vendor's HTTP 400 `errorMsg`. The vendor returns 400 for
28# a missing key, an invalid key and an exhausted balance alike; these keyword sets are the
29# only signal available and must be kept in step with observed real responses.
30QUOTA_KEYWORDS = ("balance", "credit", "quota", "insufficient", "exhaust", "limit reached")
31AUTH_KEYWORDS = ("api key", "apikey", "api_key", "unauthori", "invalid key")
34def _env_float(name: str, default: float) -> float:
35 raw = os.environ.get(name)
36 return float(raw) if raw not in (None, "") else default
39def _env_int(name: str, default: int) -> int:
40 raw = os.environ.get(name)
41 return int(raw) if raw not in (None, "") else default
44class ExternalAPIProvider:
45 """
46 Production provider that calls external verification APIs.
47 Uses landlineremover.com which returns both line type and DNC status in a single call.
48 """
50 def __init__(
51 self,
52 timeout: float = 10.0,
53 *,
54 max_retries: int | None = None,
55 backoff_base: float = 0.5,
56 backoff_max: float = 8.0,
57 retry_after_cap: float | None = None,
58 rate_limit_per_second: float | None = None,
59 breaker_failure_threshold: int | None = None,
60 breaker_reset_seconds: float = 30.0,
61 pool_size: int = 20,
62 transport: httpx.BaseTransport | None = None,
63 ):
64 """
65 Initialize external API provider.
67 Resilience settings default from the environment so deployments can tune them
68 without a code change:
69 LLS_PROVIDER_MAX_RETRIES (default 3): retries on 429/5xx/transport errors
70 LLS_PROVIDER_RETRY_AFTER_CAP_SECONDS (default 60): longest wait a vendor
71 Retry-After header can impose; set it below the caller's own time
72 budget (an API Gateway integration has 30 s in total)
73 LLS_PROVIDER_RATE_LIMIT_PER_SECOND (default 0 = off): client-side ceiling
74 LLS_PROVIDER_BREAKER_THRESHOLD (default 5): consecutive failed calls (after
75 retries) that open the circuit for breaker_reset_seconds
77 The limiter and breaker are process-local: they bound one Lambda invocation or
78 worker pool, not the fleet.
80 Args:
81 timeout: HTTP request timeout in seconds
82 max_retries: bounded retry count; backoff is exponential with full jitter
83 backoff_base: first backoff ceiling in seconds; doubles per attempt
84 backoff_max: cap on a single backoff wait
85 retry_after_cap: cap on a wait requested by a Retry-After header
86 rate_limit_per_second: token-bucket rate; burst equals the rate
87 breaker_failure_threshold: consecutive failures that open the circuit
88 breaker_reset_seconds: how long the circuit stays open before one probe call
89 pool_size: httpx connection pool size, sized for parallel workers
90 transport: httpx transport override for tests (never call the vendor in tests)
91 """
92 self.api_key = os.environ.get("LANDLINE_REMOVER_API_KEY", "")
93 if not self.api_key:
94 logger.warning("LANDLINE_REMOVER_API_KEY not set")
96 self.api_url = "https://app.landlineremover.com/api/check-number"
97 # Trailing slash is required: without it the vendor 301s to the slashed
98 # form and returns an empty body to clients that do not follow redirects.
99 self.credits_url = "https://app.landlineremover.com/api/check-credits/"
100 self.timeout = timeout
101 self.max_retries = (
102 max_retries if max_retries is not None else _env_int("LLS_PROVIDER_MAX_RETRIES", 3)
103 )
104 self.backoff_base = backoff_base
105 self.backoff_max = backoff_max
106 self.retry_after_cap = (
107 retry_after_cap
108 if retry_after_cap is not None
109 else _env_float("LLS_PROVIDER_RETRY_AFTER_CAP_SECONDS", RETRY_AFTER_CAP_SECONDS)
110 )
111 self.rate_limit_per_second = (
112 rate_limit_per_second
113 if rate_limit_per_second is not None
114 else _env_float("LLS_PROVIDER_RATE_LIMIT_PER_SECOND", 0.0)
115 )
116 self.breaker_failure_threshold = (
117 breaker_failure_threshold
118 if breaker_failure_threshold is not None
119 else _env_int("LLS_PROVIDER_BREAKER_THRESHOLD", 5)
120 )
121 self.breaker_reset_seconds = breaker_reset_seconds
122 self.pool_limits = httpx.Limits(
123 max_connections=pool_size, max_keepalive_connections=pool_size
124 )
125 self.http_client = httpx.Client(
126 timeout=timeout, limits=self.pool_limits, transport=transport
127 )
129 # Injectable for tests; never patch time globally
130 self.sleep: Callable[[float], None] = time.sleep
131 self.clock: Callable[[], float] = time.monotonic
133 self._lock = threading.Lock()
134 self._tokens = float(self.rate_limit_per_second)
135 self._last_refill = self.clock()
136 self._consecutive_failures = 0
137 self._breaker_opened_at: float | None = None
138 self._breaker_probe_in_flight = False
140 # ---- rate limiter -------------------------------------------------------------
142 def _acquire_token(self) -> None:
143 """Token bucket: wait until a request token is available (no-op when disabled)."""
144 if self.rate_limit_per_second <= 0:
145 return
146 while True:
147 with self._lock:
148 now = self.clock()
149 elapsed = max(0.0, now - self._last_refill)
150 self._tokens = min(
151 float(self.rate_limit_per_second),
152 self._tokens + elapsed * self.rate_limit_per_second,
153 )
154 self._last_refill = now
155 if self._tokens >= 1.0:
156 self._tokens -= 1.0
157 return
158 wait = (1.0 - self._tokens) / self.rate_limit_per_second
159 self.sleep(wait)
161 # ---- circuit breaker ----------------------------------------------------------
163 def _check_breaker(self) -> None:
164 """Fail fast while the circuit is open; allow a single probe once the reset elapses."""
165 with self._lock:
166 if self._breaker_opened_at is None:
167 return
168 elapsed = self.clock() - self._breaker_opened_at
169 if elapsed >= self.breaker_reset_seconds and not self._breaker_probe_in_flight:
170 self._breaker_probe_in_flight = True # half-open: one call may try
171 return
172 raise ProviderUnavailableError(
173 f"Provider circuit open after {self._consecutive_failures} consecutive failures; "
174 f"retry in {max(0.0, self.breaker_reset_seconds - elapsed):.0f}s"
175 )
177 def _record_success(self) -> None:
178 with self._lock:
179 self._consecutive_failures = 0
180 self._breaker_opened_at = None
181 self._breaker_probe_in_flight = False
183 def _record_failure(self) -> None:
184 with self._lock:
185 self._consecutive_failures += 1
186 self._breaker_probe_in_flight = False
187 if self._consecutive_failures >= self.breaker_failure_threshold:
188 if self._breaker_opened_at is None:
189 logger.error(
190 f"Provider circuit opened after {self._consecutive_failures} "
191 f"consecutive failures"
192 )
193 self._breaker_opened_at = self.clock()
195 # ---- retry --------------------------------------------------------------------
197 def _backoff(self, attempt: int, response: httpx.Response | None) -> float:
198 """Seconds to wait before retry `attempt` (0-based); honours Retry-After."""
199 if response is not None:
200 retry_after = response.headers.get("Retry-After")
201 if retry_after:
202 try:
203 return min(float(retry_after), self.retry_after_cap)
204 except ValueError:
205 pass # HTTP-date form: fall through to computed backoff
206 ceiling = min(self.backoff_base * (2**attempt), self.backoff_max)
207 return random.uniform(0, ceiling) # noqa: S311 - jitter, not security
209 def _classify_400(self, response: httpx.Response) -> ProviderError:
210 """Turn the vendor's catch-all 400 into a specific exception where the body allows."""
211 message = ""
212 try:
213 body = response.json()
214 if isinstance(body, dict):
215 message = str(body.get("errorMsg") or body.get("error") or "")
216 except ValueError:
217 message = response.text
218 lowered = message.lower()
219 if any(k in lowered for k in QUOTA_KEYWORDS):
220 return ProviderQuotaError(f"Provider quota exhausted: {message}", status_code=400)
221 if any(k in lowered for k in AUTH_KEYWORDS):
222 return ProviderAuthError(f"Provider rejected API key: {message}", status_code=400)
223 return ProviderError(f"API request failed with status 400: {message}", status_code=400)
225 def _request(self, phone: str) -> httpx.Response:
226 """GET with bounded retry on 429/5xx/transport errors. Raises ProviderError."""
227 attempt = 0
228 while True:
229 self._acquire_token()
230 response: httpx.Response | None = None
231 error: ProviderError
232 try:
233 response = self.http_client.get(
234 self.api_url,
235 params={"apikey": self.api_key, "number": phone},
236 follow_redirects=True,
237 )
238 if response.status_code < 400:
239 return response
240 if response.status_code == 400:
241 raise self._classify_400(response)
242 if response.status_code not in RETRYABLE_STATUSES:
243 logger.error(f"API error: {response.status_code} - {response.text}")
244 raise ProviderError(
245 f"API request failed with status {response.status_code}",
246 status_code=response.status_code,
247 )
248 error = ProviderError(
249 f"API request failed with status {response.status_code}",
250 status_code=response.status_code,
251 )
252 except httpx.RequestError as e:
253 error = ProviderError(f"Network error during API call: {str(e)}")
255 if attempt >= self.max_retries:
256 logger.error(f"Provider call failed after {attempt + 1} attempts: {error}")
257 raise error
258 wait = self._backoff(attempt, response)
259 logger.warning(
260 f"Provider call failed ({error}); retry {attempt + 1}/{self.max_retries} "
261 f"in {wait:.2f}s"
262 )
263 self.sleep(wait)
264 attempt += 1
266 def verify_phone(self, phone: str) -> VerificationResult:
267 """
268 Verify phone using landlineremover.com API.
270 This API returns line type, DNC status, and litigator status in a
271 single call, which is more efficient than making separate API calls.
273 Args:
274 phone: E.164 formatted phone number
276 Returns:
277 VerificationResult of (line_type, dnc, known_litigator)
279 Raises:
280 httpx.HTTPError: For API communication errors
281 ValueError: For invalid responses
282 """
283 logger.debug(f"Verifying phone {phone[:6]}*** via external API")
285 if not self.api_key:
286 # A provider configuration failure, not a data error: callers must not
287 # classify it as an invalid customer phone number.
288 raise ProviderError("API key not configured")
290 self._check_breaker()
292 try:
293 # Single call returns line type, DNC and litigator status; retried on
294 # transient failures inside _request
295 response = self._request(phone)
297 # Parse response
298 json_response = response.json()
300 # Extract data from response wrapper
301 if "data" in json_response:
302 data = json_response["data"]
303 else:
304 data = json_response
306 # Map line type from API response
307 line_type = self._map_line_type(data)
309 # Map DNC status - API uses "DNCType" field
310 # Values can be "dnc", "clean", "litigator", etc.
311 dnc_type = data.get("DNCType", data.get("dnc_type", "")).lower()
312 is_dnc = dnc_type != "clean" and dnc_type != ""
314 # Litigator status is signaled via DNCType rather than a separate
315 # field; substring match tolerates combined variants like "dnc_litigator"
316 known_litigator = "litigator" in dnc_type
318 logger.debug(
319 f"Verification complete for {phone[:6]}***",
320 extra={
321 "line_type": line_type.value,
322 "is_dnc": is_dnc,
323 "dnc_type": dnc_type,
324 "known_litigator": known_litigator,
325 },
326 )
328 self._record_success()
329 return VerificationResult(line_type, is_dnc, known_litigator)
331 except (ProviderQuotaError, ProviderAuthError):
332 # The vendor answered; the account is the problem. Not an outage signal.
333 raise
334 except ProviderError:
335 self._record_failure()
336 raise
337 except Exception as e:
338 logger.error(f"Unexpected error during verification: {str(e)}")
339 self._record_failure()
340 raise
342 def check_credits(self) -> int:
343 """Return the vendor account's remaining credit balance.
345 This endpoint is free: it takes only the API key, has no number to bill for,
346 and repeated calls leave the balance unchanged (verified against the live
347 account). That makes it safe to poll on a schedule, so the balance can be
348 watched continuously and warned on before it reaches zero, rather than being
349 discovered when customer verifications start failing.
351 Returns:
352 Remaining credits.
354 Raises:
355 ProviderAuthError: The vendor rejected the API key.
356 ProviderError: The call failed or returned an unreadable body.
357 """
358 if not self.api_key:
359 raise ProviderError("API key not configured")
361 try:
362 response = self.http_client.get(
363 self.credits_url,
364 params={"apikey": self.api_key},
365 follow_redirects=True,
366 )
367 except httpx.RequestError as e:
368 raise ProviderError(f"Network error during credit check: {e!s}") from e
370 if response.status_code == 400:
371 raise self._classify_400(response)
372 if response.status_code >= 400:
373 raise ProviderError(
374 f"Credit check failed with status {response.status_code}",
375 status_code=response.status_code,
376 )
378 try:
379 body = response.json()
380 data = body.get("data", body)
381 return int(data["RemainingCredits"])
382 except (ValueError, KeyError, TypeError) as e:
383 raise ProviderError(f"Unreadable credit check response: {e!s}") from e
385 def _map_line_type(self, data: dict) -> LineType:
386 """
387 Map API response to LineType enum.
389 Args:
390 data: API response dictionary
392 Returns:
393 LineType enum value
394 """
395 # API uses "LineType" (capitalized) field
396 line_type_str = data.get("LineType", data.get("line_type", "")).lower()
398 # Map common line types
399 line_type_map = {
400 "mobile": LineType.MOBILE,
401 "landline": LineType.LANDLINE,
402 "voip": LineType.VOIP,
403 "wireless": LineType.MOBILE, # Some APIs return "wireless" for mobile
404 "fixed": LineType.LANDLINE, # Some APIs return "fixed" for landline
405 }
407 return line_type_map.get(line_type_str, LineType.UNKNOWN)
409 def __del__(self) -> None:
410 """Cleanup HTTP client"""
411 if hasattr(self, "http_client"):
412 self.http_client.close()