"""prospectapis: the official Python client for the ProspectAPIs REST API (https://prospectapis.com/docs). Standard library only. One method per route. Every call sends your key as a Bearer token, retries a 429 or 503 once after the Retry-After the API names (at most 10 seconds), and raises ProspectApisError with the API's status, code and message. Pass idempotency_key= on any POST to make a retry safe: the API answers a repeat from the first call, unbilled. from prospectapis import ProspectApis client = ProspectApis() # reads PROSPECTAPIS_API_KEY client.verify_emails(["jane@acme.com"], depth="quick") """ import json import os import time import urllib.error import urllib.parse import urllib.request __version__ = "0.1.0" DEFAULT_BASE_URL = "https://api.prospectapis.com" class ProspectApisError(Exception): def __init__(self, status, body): body = body if isinstance(body, dict) else {} super().__init__(body.get("error") or body.get("message") or f"HTTP {status}") self.status = status self.code = body.get("code") self.body = body class ProspectApis: def __init__(self, api_key=None, base_url=DEFAULT_BASE_URL, timeout=60, opener=None): self.api_key = api_key or os.environ.get("PROSPECTAPIS_API_KEY") if not self.api_key: raise ValueError("ProspectApis needs an API key: pass api_key= or set PROSPECTAPIS_API_KEY.") self.base_url = base_url.rstrip("/") self.timeout = timeout self._open = opener or urllib.request.urlopen def request(self, method, path, query=None, body=None, idempotency_key=None, raw=False): qs = "" if query: pairs = {k: v for k, v in query.items() if v is not None} qs = "?" + urllib.parse.urlencode(pairs) if pairs else "" headers = {"Authorization": f"Bearer {self.api_key}", "Accept": "text/csv" if raw else "application/json", "User-Agent": f"prospectapis-python/{__version__}"} data = None if body is not None: headers["Content-Type"] = "application/json" data = json.dumps(body).encode() if idempotency_key: headers["Idempotency-Key"] = idempotency_key for attempt in range(2): req = urllib.request.Request(f"{self.base_url}{path}{qs}", data=data, headers=headers, method=method) try: with self._open(req, timeout=self.timeout) as res: text = res.read().decode() if raw: return text return json.loads(text) if text else None except urllib.error.HTTPError as err: if err.code in (429, 503) and attempt == 0: try: wait = float(err.headers.get("Retry-After") or 2) except ValueError: wait = 2 time.sleep(min(wait, 10)) continue text = err.read().decode(errors="replace") try: parsed = json.loads(text) if text else {} except ValueError: parsed = {"error": text[:200]} raise ProspectApisError(err.code, parsed) from None # Email verification def verify_emails(self, emails, depth=None, idempotency_key=None): body = {"emails": emails} if depth: body["depth"] = depth return self.request("POST", "/v1/email/verify", body=body, idempotency_key=idempotency_key) def find_email(self, first_name, domain, last_name=None, idempotency_key=None): body = {"first_name": first_name, "domain": domain} if last_name: body["last_name"] = last_name return self.request("POST", "/v1/email/find", body=body, idempotency_key=idempotency_key) def check_domain(self, domain): return self.request("GET", "/v1/email/domain/" + urllib.parse.quote(domain, safe="")) def report_outcomes(self, outcomes): return self.request("POST", "/v1/email/outcomes", body={"outcomes": outcomes}) def verify_stats(self, days=None): return self.request("GET", "/v1/email/verify/stats", query={"days": days}) # Bulk verification jobs def clean_list(self, emails, idempotency_key=None): return self.request("POST", "/v1/email/clean", body={"emails": emails}, idempotency_key=idempotency_key) def create_verify_job(self, emails, callback_url=None, sample=None, monitor_days=None, idempotency_key=None): body = {"emails": emails} if callback_url: body["callback_url"] = callback_url if sample is not None: body["sample"] = sample if monitor_days is not None: body["monitor_days"] = monitor_days return self.request("POST", "/v1/email/verify/jobs", body=body, idempotency_key=idempotency_key) def stop_verify_job(self, job_id): return self.request("POST", "/v1/email/verify/jobs/" + urllib.parse.quote(job_id, safe="") + "/stop") def get_verify_job(self, job_id): return self.request("GET", "/v1/email/verify/jobs/" + urllib.parse.quote(job_id, safe="")) def verify_job_results(self, job_id, cursor=None, limit=None, format=None): return self.request("GET", "/v1/email/verify/jobs/" + urllib.parse.quote(job_id, safe="") + "/results", query={"cursor": cursor, "limit": limit, "format": format}, raw=format == "csv") def iter_verify_job_results(self, job_id): """Every result of a job, page by page.""" cursor = 0 while cursor is not None: page = self.verify_job_results(job_id, cursor=cursor) yield from page["data"] cursor = page.get("next_cursor") def wait_for_verify_job(self, job_id, interval=15, timeout=6 * 3600): """Polls a job until it completes, then returns it.""" end = time.time() + timeout while True: job = self.get_verify_job(job_id) if job["status"] in ("completed", "cancelled"): return job if time.time() > end: raise TimeoutError(f"job {job_id} did not complete in time") time.sleep(interval) # Funding, research, watchlists, account def search_funding(self, **query): return self.request("GET", "/v1/funding", query=query) def get_funding(self, funding_id): return self.request("GET", "/v1/funding/" + urllib.parse.quote(funding_id, safe="")) def submit_research(self, body, idempotency_key=None): return self.request("POST", "/v1/research", body=body, idempotency_key=idempotency_key) def get_research(self, research_id): return self.request("GET", "/v1/research/" + urllib.parse.quote(research_id, safe="")) def create_watchlist(self, body): return self.request("POST", "/v1/watchlists", body=body) def list_watchlists(self): return self.request("GET", "/v1/watchlists") def get_watchlist(self, watchlist_id): return self.request("GET", "/v1/watchlists/" + urllib.parse.quote(watchlist_id, safe="")) def delete_watchlist(self, watchlist_id): return self.request("DELETE", "/v1/watchlists/" + urllib.parse.quote(watchlist_id, safe="")) def account(self): return self.request("GET", "/v1/account") def pricing(self): return self.request("GET", "/pricing") __all__ = ["ProspectApis", "ProspectApisError", "DEFAULT_BASE_URL", "__version__"]