From 7632fff8011b8a2224df487d91eacd619b3d787d Mon Sep 17 00:00:00 2001 From: Francisc Balint <7178547+franciscbalint@users.noreply.github.com> Date: Sun, 18 Oct 2026 00:50:35 +0400 Subject: [PATCH 1/7] feat: v2 validate parity (verify_plus, activity_data, timeouts) Add optional validate and validate_batch parameters and activity fields on validation responses per ZeroBounce v2 docs. Co-authored-by: Cursor --- src/zerobouncesdk/zb_validate_response.py | 6 ++ src/zerobouncesdk/zerobouncesdk.py | 68 ++++++++++++++++++++--- tests/zero_bounce_test_case.py | 34 ++++++++++++ 3 files changed, 100 insertions(+), 8 deletions(-) diff --git a/src/zerobouncesdk/zb_validate_response.py b/src/zerobouncesdk/zb_validate_response.py index 04a78c7..c8c8789 100644 --- a/src/zerobouncesdk/zb_validate_response.py +++ b/src/zerobouncesdk/zb_validate_response.py @@ -34,6 +34,12 @@ class ZBValidateResponse(ZBResponse): domain_age_days: str = None """Age of the email domain in days or [null].""" + active_in_days: str = None + """Last activity bucket when activity_data was requested (e.g. 30, 365+).""" + + active_first_seen: str = None + """First date activity was detected (YYYY-MM-DD).""" + free_email: bool = False """[true/false] If the email comes from a free provider.""" diff --git a/src/zerobouncesdk/zerobouncesdk.py b/src/zerobouncesdk/zerobouncesdk.py index 9bb88a1..888598c 100644 --- a/src/zerobouncesdk/zerobouncesdk.py +++ b/src/zerobouncesdk/zerobouncesdk.py @@ -232,7 +232,24 @@ def get_activity(self, email: str): f"{self._base_url}/activity", ZBGetActivityResponse, params={"email": email} ) - def validate(self, email: str, ip_address: str = None): + @staticmethod + def _check_validate_timeout(timeout: Optional[int]) -> None: + if timeout is not None and (timeout < 3 or timeout > 60): + raise ZBClientException("Invalid parameter: timeout (must be between 3 and 60 seconds)") + + @staticmethod + def _check_validate_batch_timeout(timeout: Optional[int]) -> None: + if timeout is not None and (timeout < 10 or timeout > 120): + raise ZBClientException("Invalid parameter: timeout (must be between 10 and 120 seconds)") + + def validate( + self, + email: str, + ip_address: str = None, + timeout: Optional[int] = None, + activity_data: Optional[bool] = None, + verify_plus: Optional[bool] = None, + ): """Validates the given email address. Parameters @@ -241,33 +258,61 @@ def validate(self, email: str, ip_address: str = None): The email address you want to validate ip_address: str or None The IP Address the email signed up from (Can be blank) + timeout: int or None + Validation timeout in seconds (3-60). Optional. + activity_data: bool or None + When set, sends ``activity_data`` query parameter. + verify_plus: bool or None + When set, sends ``verify_plus`` query parameter. Raises ------ ZBApiException + ZBClientException Returns ------- response: ZBValidateResponse Returns a ZBValidateResponse object if the request was successful """ + self._check_validate_timeout(timeout) + + params = { + "email": email, + "ip_address": ip_address, + } + if timeout is not None: + params["timeout"] = timeout + if activity_data is not None: + params["activity_data"] = "true" if activity_data else "false" + if verify_plus is not None: + params["verify_plus"] = "true" if verify_plus else "false" return self._get( f"{self._base_url}/validate", ZBValidateResponse, - params={ - "email": email, - "ip_address": ip_address, - }, + params=params, ) - def validate_batch(self, email_batch: List[ZBValidateBatchElement]): + def validate_batch( + self, + email_batch: List[ZBValidateBatchElement], + timeout: Optional[int] = None, + activity_data: Optional[bool] = None, + verify_plus: Optional[bool] = None, + ): """Allows you to send us batches up to 100 emails at a time. Parameters ---------- email_batch: List[ZBValidateBatchElement] Array of ZBValidateBatchElement + timeout: int or None + Batch validation timeout in seconds (10-120). Optional. + activity_data: bool or None + When set, includes activity data on each result. + verify_plus: bool or None + When true, uses Verify+ for validation. Raises ------ @@ -281,17 +326,24 @@ def validate_batch(self, email_batch: List[ZBValidateBatchElement]): """ if not email_batch: raise ZBClientException("Empty parameter: email_batch") + self._check_validate_batch_timeout(timeout) - json={ + payload = { "api_key": self._api_key, "email_batch": [ batch_element.to_json() for batch_element in email_batch ], } + if timeout is not None: + payload["timeout"] = timeout + if activity_data is not None: + payload["activity_data"] = activity_data + if verify_plus is not None: + payload["verify_plus"] = verify_plus return self._post( f"{self._base_url}/validatebatch", ZBValidateBatchResponse, - json=json) + json=payload) def _send_file( self, diff --git a/tests/zero_bounce_test_case.py b/tests/zero_bounce_test_case.py index 47e2db1..9a1a8dd 100644 --- a/tests/zero_bounce_test_case.py +++ b/tests/zero_bounce_test_case.py @@ -112,6 +112,23 @@ def test_validate_valid(self): self.assertEqual(response.sub_status, ZBValidateSubStatus.mailbox_not_found) self.assertEqual(response.processed_at, datetime(2023, 3, 28, 12, 30, 18, 990000)) + def test_validate_sends_optional_query_params(self): + self.requests_mock.get.return_value = MockResponse({"address": "a@b.com", "status": "valid"}) + self.zero_bounce_client.validate( + "a@b.com", + timeout=10, + activity_data=True, + verify_plus=False, + ) + params = self.requests_mock.get.call_args[1]["params"] + self.assertEqual(params["timeout"], 10) + self.assertEqual(params["activity_data"], "true") + self.assertEqual(params["verify_plus"], "false") + + def test_validate_timeout_out_of_range(self): + with self.assertRaises(ZBClientException): + self.zero_bounce_client.validate("a@b.com", timeout=2) + def test_response_sub_status_accept_all(self): self.requests_mock.get.return_value = MockResponse({ "address": "none@example.com", @@ -200,6 +217,23 @@ def test_validate_batch_valid(self): self.assertEqual(response.email_batch[1].sub_status, ZBValidateSubStatus.mailbox_not_found) self.assertEqual(len(response.errors), 0) + def test_validate_batch_sends_optional_body_fields(self): + self.requests_mock.post.return_value = MockResponse({"email_batch": [], "errors": []}) + self.zero_bounce_client.validate_batch( + [ZBValidateBatchElement("a@b.com")], + timeout=60, + activity_data=True, + verify_plus=True, + ) + payload = self.requests_mock.post.call_args[1]["json"] + self.assertEqual(payload["timeout"], 60) + self.assertTrue(payload["activity_data"]) + self.assertTrue(payload["verify_plus"]) + + def test_validate_batch_timeout_out_of_range(self): + with self.assertRaises(ZBClientException): + self.zero_bounce_client.validate_batch([ZBValidateBatchElement("a@b.com")], timeout=9) + def test_validate_batch_uses_api_base_url(self): self.requests_mock.post.return_value = MockResponse({"email_batch": [], "errors": []}) self.zero_bounce_client.validate_batch([ZBValidateBatchElement("a@b.com")]) From f8c546f7872f9e656ee5fab9d376aee13af7c5a7 Mon Sep 17 00:00:00 2001 From: Francisc Balint <7178547+franciscbalint@users.noreply.github.com> Date: Sun, 18 Oct 2026 01:46:52 +0400 Subject: [PATCH 2/7] feat: add score_email for GET/POST /v2/scoring Co-authored-by: Cursor --- src/zerobouncesdk/__init__.py | 1 + src/zerobouncesdk/zb_score_email_response.py | 8 +++ src/zerobouncesdk/zerobouncesdk.py | 52 ++++++++++++++++++++ tests/zero_bounce_test_case.py | 39 +++++++++++++++ 4 files changed, 100 insertions(+) create mode 100644 src/zerobouncesdk/zb_score_email_response.py diff --git a/src/zerobouncesdk/__init__.py b/src/zerobouncesdk/__init__.py index 8c40c56..3e1f11a 100644 --- a/src/zerobouncesdk/__init__.py +++ b/src/zerobouncesdk/__init__.py @@ -6,6 +6,7 @@ from .zb_get_credits_response import ZBGetCreditsResponse from .zb_get_api_usage_response import ZBGetApiUsageResponse from .zb_get_activity_response import ZBGetActivityResponse +from .zb_score_email_response import ZBScoreEmailResponse from .zb_validate_status import ZBValidateStatus from .zb_validate_sub_status import ZBValidateSubStatus from .zb_validate_response import ZBValidateResponse diff --git a/src/zerobouncesdk/zb_score_email_response.py b/src/zerobouncesdk/zb_score_email_response.py new file mode 100644 index 0000000..789649c --- /dev/null +++ b/src/zerobouncesdk/zb_score_email_response.py @@ -0,0 +1,8 @@ +from ._zb_response import ZBResponse + + +class ZBScoreEmailResponse(ZBResponse): + """Response for GET/POST /scoring (real-time AI scoring).""" + + email: str = None + score: int = None diff --git a/src/zerobouncesdk/zerobouncesdk.py b/src/zerobouncesdk/zerobouncesdk.py index 888598c..21db73a 100644 --- a/src/zerobouncesdk/zerobouncesdk.py +++ b/src/zerobouncesdk/zerobouncesdk.py @@ -16,6 +16,7 @@ ZBGetCreditsResponse, ZBGetApiUsageResponse, ZBGetActivityResponse, + ZBScoreEmailResponse, ZBValidateResponse, ZBValidateBatchElement, ZBValidateBatchResponse, @@ -165,6 +166,26 @@ def _post(self, url, response_class, data=None, json=None, files=None): return response_class(json_response) + def _post_form(self, url, response_class, data=None): + if not data: + data = {} + data = dict(data) + data["api_key"] = self._api_key + try: + response = requests.post(url, data=data, timeout=self._timeout_s) + except RequestException as e: + raise type(e)(self._redact(e)) from None + + try: + json_response = response.json() + except ValueError as e: + raise ZBApiException from e + + error = json_response.pop("error", None) + if error: + raise ZBApiException(error) + return response_class(json_response) + def get_credits(self): """Tells you how many credits you have left on your account. It's simple, fast and easy to use. @@ -232,6 +253,37 @@ def get_activity(self, email: str): f"{self._base_url}/activity", ZBGetActivityResponse, params={"email": email} ) + def score_email(self, email: str, use_post: bool = False): + """Real-time AI scoring for a single email (GET or POST /scoring). + + Parameters + ---------- + email: str + The email address to score + use_post: bool + When True, sends ``api_key`` and ``email`` as form-encoded POST body. + + Raises + ------ + ZBApiException + + Returns + ------- + response: ZBScoreEmailResponse + """ + + if use_post: + return self._post_form( + f"{self._base_url}/scoring", + ZBScoreEmailResponse, + data={"email": email}, + ) + return self._get( + f"{self._base_url}/scoring", + ZBScoreEmailResponse, + params={"email": email}, + ) + @staticmethod def _check_validate_timeout(timeout: Optional[int]) -> None: if timeout is not None and (timeout < 3 or timeout > 60): diff --git a/tests/zero_bounce_test_case.py b/tests/zero_bounce_test_case.py index 9a1a8dd..279e295 100644 --- a/tests/zero_bounce_test_case.py +++ b/tests/zero_bounce_test_case.py @@ -98,6 +98,45 @@ def test_activity_valid(self): self.assertTrue(response.found) self.assertEqual(response.active_in_days, "180") + def test_score_email_get(self): + self.requests_mock.get.return_value = MockResponse({ + "email": "mark@gmail.com", + "score": 10, + }) + + response = self.zero_bounce_client.score_email("mark@gmail.com") + self.assertEqual(response.email, "mark@gmail.com") + self.assertEqual(response.score, 10) + self.requests_mock.get.assert_called_with( + f"{self.zero_bounce_client._base_url}/scoring", + params={"api_key": "dummy_key", "email": "mark@gmail.com"}, + timeout=120.0, + ) + + def test_score_email_post(self): + self.requests_mock.post.return_value = MockResponse({ + "email": "mark@gmail.com", + "score": 10, + }) + + response = self.zero_bounce_client.score_email("mark@gmail.com", use_post=True) + self.assertEqual(response.email, "mark@gmail.com") + self.assertEqual(response.score, 10) + self.requests_mock.post.assert_called_with( + f"{self.zero_bounce_client._base_url}/scoring", + data={"api_key": "dummy_key", "email": "mark@gmail.com"}, + timeout=120.0, + ) + + def test_score_email_api_error(self): + self.requests_mock.get.return_value = MockResponse({ + "error": "Please provide a valid email address", + }) + + with self.assertRaises(ZBApiException) as cm: + self.zero_bounce_client.score_email("not-an-email") + self.assertEqual(str(cm.exception), "Please provide a valid email address") + def test_validate_valid(self): self.requests_mock.get.return_value = MockResponse({ "address": "invalid@example.com", From 317f7d64e0a08d01c4c57f1b38bd47fabd491e81 Mon Sep 17 00:00:00 2001 From: Francisc Balint <7178547+franciscbalint@users.noreply.github.com> Date: Sun, 18 Oct 2026 02:34:21 +0400 Subject: [PATCH 3/7] feat: Priority 3 API parity (bulk finder, filters, S3) Mirror JavaScript P3 surfaces: email-finder and domain-search bulk jobs, list evaluator, allow/block filters, and S3 validate/score/export helpers. Co-authored-by: Cursor --- src/zerobouncesdk/zerobouncesdk.py | 385 +++++++++++++++++++++++++++++ tests/zero_bounce_test_case.py | 39 +++ 2 files changed, 424 insertions(+) diff --git a/src/zerobouncesdk/zerobouncesdk.py b/src/zerobouncesdk/zerobouncesdk.py index 21db73a..8f93c0e 100644 --- a/src/zerobouncesdk/zerobouncesdk.py +++ b/src/zerobouncesdk/zerobouncesdk.py @@ -34,6 +34,9 @@ class ZeroBounce: """The ZeroBounce main class. All the requests are implemented here.""" BULK_BASE_URL = "https://bulkapi.zerobounce.net/v2" + BULK_ROOT_URL = "https://bulkapi.zerobounce.net" + EMAIL_FINDER_BASE_URL = "https://bulkapi.zerobounce.net/email-finder" + DOMAIN_SEARCH_BASE_URL = "https://bulkapi.zerobounce.net/domain-search" SCORING_BASE_URL = "https://bulkapi.zerobounce.net/v2/scoring" DEFAULT_HTTP_TIMEOUT_S = 120.0 @@ -186,6 +189,107 @@ def _post_form(self, url, response_class, data=None): raise ZBApiException(error) return response_class(json_response) + def _request_json(self, method: str, url: str, *, params=None, data=None, files=None): + if params is None: + params = {} + if "api_key" not in params and (data is None or "api_key" not in data): + params = dict(params) + params["api_key"] = self._api_key + try: + response = requests.request( + method, + url, + params=params, + data=data, + files=files, + timeout=self._timeout_s, + ) + except RequestException as e: + raise type(e)(self._redact(e)) from None + try: + return response.json() + except ValueError as e: + raise ZBApiException( + f"Request not processed successfully. Status code {response.status_code}" + ) from e + + def _send_bulk_root_multipart( + self, url: str, file_path: str, fields: dict + ) -> ZBSendFileResponse: + data = dict(fields) + data["api_key"] = self._api_key + with open(file_path, "rb") as file: + files = {"file": (os.path.basename(file_path), file, "text/csv")} + return self._post(url, ZBSendFileResponse, data=data, files=files) + + def _send_bulk_root_multipart_stream( + self, url: str, file_stream: BinaryIO, file_name: str, fields: dict + ) -> ZBSendFileResponse: + data = dict(fields) + data["api_key"] = self._api_key + files = {"file": (file_name, file_stream, "text/csv")} + return self._post(url, ZBSendFileResponse, data=data, files=files) + + def _bulk_root_file_status(self, base_url: str, file_id: str) -> ZBFileStatusResponse: + if not file_id.strip(): + raise ZBClientException("Empty parameter: file_id") + return self._get( + f"{base_url}/filestatus", + ZBFileStatusResponse, + params={"file_id": file_id}, + ) + + def _bulk_root_get_file( + self, + base_url: str, + file_id: str, + download_path: str, + ) -> ZBGetFileResponse: + if not file_id.strip(): + raise ZBClientException("Empty parameter: file_id") + params = {"api_key": self._api_key, "file_id": file_id} + try: + response = requests.get( + f"{base_url}/getfile", + params=params, + timeout=self._timeout_s, + ) + except RequestException as e: + raise type(e)(self._redact(e)) from None + body_bytes = response.content or b"" + body_str = body_bytes.decode("utf-8", errors="replace") + content_type = response.headers.get("Content-Type") or "" + if response.status_code > 299: + if body_str.lstrip().startswith("{"): + raise ZBApiException(self._redact(self._format_get_file_error_message(body_str))) + raise ZBApiException(self._redact(body_str or f"HTTP {response.status_code}")) + if self._should_treat_get_file_body_as_error(body_str, content_type): + raise ZBApiException(self._redact(self._format_get_file_error_message(body_str))) + dirname = os.path.dirname(download_path) + if dirname: + os.makedirs(dirname, exist_ok=True) + with open(download_path, "wb") as f: + f.write(body_bytes) + return ZBGetFileResponse({"local_file_path": download_path}) + + def _bulk_root_delete_file(self, base_url: str, file_id: str) -> ZBDeleteFileResponse: + if not file_id.strip(): + raise ZBClientException("Empty parameter: file_id") + return self._get( + f"{base_url}/deletefile", + ZBDeleteFileResponse, + params={"file_id": file_id}, + ) + + def _post_bulk_root_form(self, path: str, fields: dict) -> ZBSendFileResponse: + data = dict(fields) + data["api_key"] = self._api_key + return self._post( + f"{self.BULK_ROOT_URL}{path}", + ZBSendFileResponse, + data=data, + ) + def get_credits(self): """Tells you how many credits you have left on your account. It's simple, fast and easy to use. @@ -863,6 +967,287 @@ def scoring_delete_file(self, file_id: str): return self._delete_file(True, file_id) + # Priority 3 — bulk email finder (bulkapi …/email-finder/*) + + def send_email_finder_file( + self, + file_path: str, + domain_column: int, + first_name_column: int = None, + last_name_column: int = None, + middle_name_column: int = None, + full_name_column: int = None, + has_header_row: bool = False, + ) -> ZBSendFileResponse: + if not domain_column: + raise ZBClientException("Empty parameter: domain_column") + fields = {"domain_column": domain_column, "has_header_row": has_header_row} + if first_name_column is not None: + fields["first_name_column"] = first_name_column + if last_name_column is not None: + fields["last_name_column"] = last_name_column + if middle_name_column is not None: + fields["middle_name_column"] = middle_name_column + if full_name_column is not None: + fields["full_name_column"] = full_name_column + return self._send_bulk_root_multipart( + f"{self.EMAIL_FINDER_BASE_URL}/sendfile", file_path, fields + ) + + def send_email_finder_file_stream( + self, + file_stream: BinaryIO, + file_name: str, + domain_column: int, + first_name_column: int = None, + last_name_column: int = None, + middle_name_column: int = None, + full_name_column: int = None, + has_header_row: bool = False, + ) -> ZBSendFileResponse: + if not domain_column: + raise ZBClientException("Empty parameter: domain_column") + fields = {"domain_column": domain_column, "has_header_row": has_header_row} + if first_name_column is not None: + fields["first_name_column"] = first_name_column + if last_name_column is not None: + fields["last_name_column"] = last_name_column + if middle_name_column is not None: + fields["middle_name_column"] = middle_name_column + if full_name_column is not None: + fields["full_name_column"] = full_name_column + return self._send_bulk_root_multipart_stream( + f"{self.EMAIL_FINDER_BASE_URL}/sendfile", + file_stream, + file_name, + fields, + ) + + def email_finder_file_status(self, file_id: str) -> ZBFileStatusResponse: + return self._bulk_root_file_status(self.EMAIL_FINDER_BASE_URL, file_id) + + def email_finder_get_file(self, file_id: str, download_path: str) -> ZBGetFileResponse: + return self._bulk_root_get_file(self.EMAIL_FINDER_BASE_URL, file_id, download_path) + + def email_finder_delete_file(self, file_id: str) -> ZBDeleteFileResponse: + return self._bulk_root_delete_file(self.EMAIL_FINDER_BASE_URL, file_id) + + # Priority 3 — bulk domain search (bulkapi …/domain-search/*) + + def send_domain_search_file( + self, + file_path: str, + domain_column: int, + has_header_row: bool = False, + ) -> ZBSendFileResponse: + if not domain_column: + raise ZBClientException("Empty parameter: domain_column") + return self._send_bulk_root_multipart( + f"{self.DOMAIN_SEARCH_BASE_URL}/sendfile", + file_path, + {"domain_column": domain_column, "has_header_row": has_header_row}, + ) + + def send_domain_search_file_stream( + self, + file_stream: BinaryIO, + file_name: str, + domain_column: int, + has_header_row: bool = False, + ) -> ZBSendFileResponse: + if not domain_column: + raise ZBClientException("Empty parameter: domain_column") + return self._send_bulk_root_multipart_stream( + f"{self.DOMAIN_SEARCH_BASE_URL}/sendfile", + file_stream, + file_name, + {"domain_column": domain_column, "has_header_row": has_header_row}, + ) + + def domain_search_file_status(self, file_id: str) -> ZBFileStatusResponse: + return self._bulk_root_file_status(self.DOMAIN_SEARCH_BASE_URL, file_id) + + def domain_search_get_file(self, file_id: str, download_path: str) -> ZBGetFileResponse: + return self._bulk_root_get_file(self.DOMAIN_SEARCH_BASE_URL, file_id, download_path) + + def domain_search_delete_file(self, file_id: str) -> ZBDeleteFileResponse: + return self._bulk_root_delete_file(self.DOMAIN_SEARCH_BASE_URL, file_id) + + # Priority 3 — list evaluator + + def send_list_evaluator_file( + self, file_path: str, email_address_column: int + ) -> ZBSendFileResponse: + if not email_address_column: + raise ZBClientException("Empty parameter: email_address_column") + data = { + "api_key": self._api_key, + "email_address_column": email_address_column, + } + with open(file_path, "rb") as file: + files = {"file": (os.path.basename(file_path), file, "text/csv")} + return self._post( + f"{self.BULK_BASE_URL}/listevaluator/", + ZBSendFileResponse, + data=data, + files=files, + ) + + def send_list_evaluator_file_stream( + self, + file_stream: BinaryIO, + file_name: str, + email_address_column: int, + ) -> ZBSendFileResponse: + if not email_address_column: + raise ZBClientException("Empty parameter: email_address_column") + data = { + "api_key": self._api_key, + "email_address_column": email_address_column, + } + files = {"file": (file_name, file_stream, "text/csv")} + return self._post( + f"{self.BULK_BASE_URL}/listevaluator/", + ZBSendFileResponse, + data=data, + files=files, + ) + + def get_list_evaluator_result(self, file_id: str): + if not file_id.strip(): + raise ZBClientException("Empty parameter: file_id") + return self._request_json( + "GET", + f"{self.BULK_BASE_URL}/listevaluator/{file_id}/", + params={"api_key": self._api_key}, + ) + + def send_evaluated_file(self, file_path: str) -> ZBSendFileResponse: + data = {"api_key": self._api_key} + with open(file_path, "rb") as file: + files = {"file": (os.path.basename(file_path), file, "text/csv")} + return self._post( + f"{self.BULK_BASE_URL}/send-evaluated-file/", + ZBSendFileResponse, + data=data, + files=files, + ) + + # Priority 3 — custom filters (main API) + + def add_filter(self, rule: str, target: str, value: str): + if not rule or not target or not value: + raise ZBClientException("Empty parameter: rule, target, and value are required") + return self._request_json( + "POST", + f"{self._base_url}/filters/add", + data={ + "api_key": self._api_key, + "rule": rule, + "target": target, + "value": value, + }, + ) + + def list_filters(self, use_post: bool = False): + if use_post: + return self._request_json( + "POST", + f"{self._base_url}/filters/list", + data={"api_key": self._api_key}, + ) + return self._request_json( + "GET", + f"{self._base_url}/filters/list", + params={"api_key": self._api_key}, + ) + + def delete_filter(self, rule: str, target: str, value: str): + if not rule or not target or not value: + raise ZBClientException("Empty parameter: rule, target, and value are required") + return self._request_json( + "POST", + f"{self._base_url}/filters/delete", + data={ + "api_key": self._api_key, + "rule": rule, + "target": target, + "value": value, + }, + ) + + # Priority 3 — S3 bulk helpers (bulkapi root) + + def validate_s3file( + self, + file_name: str, + has_header_row: bool = None, + remove_duplicates: bool = None, + remove_duplicate: bool = None, + email_address_column: int = None, + return_url: str = None, + first_name_column: int = None, + last_name_column: int = None, + gender_column: int = None, + ip_address_column: int = None, + allow_phase_2: Optional[bool] = None, + ) -> ZBSendFileResponse: + if not file_name or not file_name.strip(): + raise ZBClientException("Empty parameter: file_name") + fields = {"file_name": file_name} + if has_header_row is not None: + fields["has_header_row"] = has_header_row + if remove_duplicates is not None: + fields["remove_duplicates"] = remove_duplicates + if remove_duplicate is not None: + fields["remove_duplicate"] = remove_duplicate + if email_address_column is not None: + fields["email_address_column"] = email_address_column + if return_url is not None: + fields["return_url"] = return_url + if first_name_column is not None: + fields["first_name_column"] = first_name_column + if last_name_column is not None: + fields["last_name_column"] = last_name_column + if gender_column is not None: + fields["gender_column"] = gender_column + if ip_address_column is not None: + fields["ip_address_column"] = ip_address_column + if allow_phase_2 is not None: + fields["allow_phase_2"] = "true" if allow_phase_2 else "false" + return self._post_bulk_root_form("/validateS3file", fields) + + def score_s3file( + self, + file_name: str, + has_header_row: bool = None, + remove_duplicates: bool = None, + remove_duplicate: bool = None, + email_address_column: int = None, + ) -> ZBSendFileResponse: + if not file_name or not file_name.strip(): + raise ZBClientException("Empty parameter: file_name") + fields = {"file_name": file_name} + if has_header_row is not None: + fields["has_header_row"] = has_header_row + if remove_duplicates is not None: + fields["remove_duplicates"] = remove_duplicates + if remove_duplicate is not None: + fields["remove_duplicate"] = remove_duplicate + if email_address_column is not None: + fields["email_address_column"] = email_address_column + return self._post_bulk_root_form("/scoreS3file", fields) + + def export_s3file(self, file_id: str, export_bucket: str) -> ZBSendFileResponse: + if not file_id.strip(): + raise ZBClientException("Empty parameter: file_id") + if not export_bucket or not export_bucket.strip(): + raise ZBClientException("Empty parameter: export_bucket") + return self._post_bulk_root_form( + "/exportS3file", + {"file_id": file_id, "export_bucket": export_bucket}, + ) + def guess_format( self, domain: str, diff --git a/tests/zero_bounce_test_case.py b/tests/zero_bounce_test_case.py index 279e295..0df11f8 100644 --- a/tests/zero_bounce_test_case.py +++ b/tests/zero_bounce_test_case.py @@ -1,4 +1,5 @@ from datetime import date, datetime, timedelta +from io import BytesIO from pathlib import Path @@ -718,3 +719,41 @@ def test_posts_pass_timeout(self): response = client._post("https://example.com", ZBResponse) self.assertEqual(response.a, 'b') self.requests_mock.post.assert_called_with('https://example.com', data=None, json=None, files=None, timeout=0.123) + + def test_send_email_finder_file_uses_bulk_email_finder_url(self): + self.requests_mock.post.return_value = MockResponse({ + "success": True, + "file_id": "ef-1", + }) + stream = BytesIO(b"domain,name\nacme.com,Jane") + self.zero_bounce_client.send_email_finder_file_stream( + stream, "finder.csv", domain_column=1 + ) + url = self.requests_mock.post.call_args[0][0] + self.assertEqual(url, "https://bulkapi.zerobounce.net/email-finder/sendfile") + + def test_add_filter_posts_to_main_api(self): + self.requests_mock.request.return_value = MockResponse( + {"Message": "Filter successfully added"} + ) + self.zero_bounce_client.add_filter("allow", "domain", "example.com") + method, url = self.requests_mock.request.call_args[0][:2] + self.assertEqual(method, "POST") + self.assertTrue(url.endswith("/filters/add")) + data = self.requests_mock.request.call_args[1]["data"] + self.assertEqual(data["rule"], "allow") + + def test_validate_s3file_posts_to_bulk_root(self): + self.requests_mock.post.return_value = MockResponse({"success": True, "file_id": "s3-1"}) + self.zero_bounce_client.validate_s3file( + "lists/jan.csv", has_header_row=True, remove_duplicates=True + ) + url = self.requests_mock.post.call_args[0][0] + self.assertEqual(url, "https://bulkapi.zerobounce.net/validateS3file") + + def test_get_list_evaluator_result_url(self): + self.requests_mock.request.return_value = MockResponse({"file_status": "Complete"}) + self.zero_bounce_client.get_list_evaluator_result("abc-123") + method, url = self.requests_mock.request.call_args[0][:2] + self.assertEqual(method, "GET") + self.assertIn("/v2/listevaluator/abc-123/", url) From d27ab4acadf0ee4c241bb951982e60d297dfc3a3 Mon Sep 17 00:00:00 2001 From: Francisc Balint <7178547+franciscbalint@users.noreply.github.com> Date: Sun, 18 Oct 2026 03:17:25 +0400 Subject: [PATCH 4/7] feat: POST /validate transport parity and listS3files Add form-encoded POST option on single validate (alongside GET) and GET listS3files on bulkapi root, with unit tests and API parity doc updates. --- src/zerobouncesdk/zerobouncesdk.py | 31 ++++++++++++++++++++++++++++++ tests/zero_bounce_test_case.py | 17 ++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/src/zerobouncesdk/zerobouncesdk.py b/src/zerobouncesdk/zerobouncesdk.py index 8f93c0e..0e15d16 100644 --- a/src/zerobouncesdk/zerobouncesdk.py +++ b/src/zerobouncesdk/zerobouncesdk.py @@ -405,6 +405,7 @@ def validate( timeout: Optional[int] = None, activity_data: Optional[bool] = None, verify_plus: Optional[bool] = None, + use_post: bool = False, ): """Validates the given email address. @@ -420,6 +421,8 @@ def validate( When set, sends ``activity_data`` query parameter. verify_plus: bool or None When set, sends ``verify_plus`` query parameter. + use_post: bool + When True, sends parameters as form-encoded POST body. Raises ------ @@ -444,6 +447,12 @@ def validate( if verify_plus is not None: params["verify_plus"] = "true" if verify_plus else "false" + if use_post: + return self._post_form( + f"{self._base_url}/validate", + ZBValidateResponse, + data=params, + ) return self._get( f"{self._base_url}/validate", ZBValidateResponse, @@ -1178,6 +1187,28 @@ def delete_filter(self, rule: str, target: str, value: str): # Priority 3 — S3 bulk helpers (bulkapi root) + def list_s3files( + self, + bucket: str = None, + search: str = None, + ): + """List S3 bucket files available for validation or scoring (GET /listS3files).""" + params = {} + if bucket is not None: + params["bucket"] = bucket + if search is not None: + params["search"] = search + payload = self._request_json( + "GET", + f"{self.BULK_ROOT_URL}/listS3files", + params=params, + ) + if isinstance(payload, dict) and payload.get("error"): + raise ZBApiException(payload["error"]) + if not isinstance(payload, list): + raise ZBApiException("Unexpected listS3files response") + return payload + def validate_s3file( self, file_name: str, diff --git a/tests/zero_bounce_test_case.py b/tests/zero_bounce_test_case.py index 0df11f8..0868557 100644 --- a/tests/zero_bounce_test_case.py +++ b/tests/zero_bounce_test_case.py @@ -165,6 +165,23 @@ def test_validate_sends_optional_query_params(self): self.assertEqual(params["activity_data"], "true") self.assertEqual(params["verify_plus"], "false") + def test_validate_post_uses_form_body(self): + self.requests_mock.post.return_value = MockResponse({"address": "a@b.com", "status": "valid"}) + self.zero_bounce_client.validate("a@b.com", use_post=True, timeout=30) + self.requests_mock.post.assert_called_once() + data = self.requests_mock.post.call_args[1]["data"] + self.assertEqual(data["email"], "a@b.com") + self.assertEqual(data["timeout"], 30) + + def test_list_s3files_gets_bulk_root(self): + self.requests_mock.get.return_value = MockResponse( + [{"name": "uploads/a.csv", "size": 10, "last_modified": "2024-01-01T00:00:00Z"}] + ) + result = self.zero_bounce_client.list_s3files(bucket="my-bucket", search="a") + self.assertEqual(result[0]["name"], "uploads/a.csv") + called_url = self.requests_mock.get.call_args[0][0] + self.assertIn("bulkapi.zerobounce.net/listS3files", called_url) + def test_validate_timeout_out_of_range(self): with self.assertRaises(ZBClientException): self.zero_bounce_client.validate("a@b.com", timeout=2) From 4b36baa4d5beb3e2ca22ce7d021c75f0f94772a0 Mon Sep 17 00:00:00 2001 From: Francisc Balint <7178547+franciscbalint@users.noreply.github.com> Date: Sun, 18 Oct 2026 14:59:37 +0400 Subject: [PATCH 5/7] fix: mock requests.request in listS3files test Co-authored-by: Cursor --- tests/zero_bounce_test_case.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/zero_bounce_test_case.py b/tests/zero_bounce_test_case.py index 0868557..e40a3d1 100644 --- a/tests/zero_bounce_test_case.py +++ b/tests/zero_bounce_test_case.py @@ -174,12 +174,12 @@ def test_validate_post_uses_form_body(self): self.assertEqual(data["timeout"], 30) def test_list_s3files_gets_bulk_root(self): - self.requests_mock.get.return_value = MockResponse( + self.requests_mock.request.return_value = MockResponse( [{"name": "uploads/a.csv", "size": 10, "last_modified": "2024-01-01T00:00:00Z"}] ) result = self.zero_bounce_client.list_s3files(bucket="my-bucket", search="a") self.assertEqual(result[0]["name"], "uploads/a.csv") - called_url = self.requests_mock.get.call_args[0][0] + called_url = self.requests_mock.request.call_args[0][1] self.assertIn("bulkapi.zerobounce.net/listS3files", called_url) def test_validate_timeout_out_of_range(self): From 2000394c9035aa43d698adbd92942d035d81e551 Mon Sep 17 00:00:00 2001 From: Francisc Balint <7178547+franciscbalint@users.noreply.github.com> Date: Mon, 19 Oct 2026 06:29:04 +0400 Subject: [PATCH 6/7] fix: return short API errors and encode form booleans HTML access-denied pages and parser exceptions were reaching callers, form flags were sent as True/False, and column 0 was rejected as empty. Co-authored-by: Cursor --- .github/workflows/publish.yml | 5 +- .github/workflows/sdk_ci.yml | 5 +- AGENTS.md | 57 ++ CHANGELOG.md | 140 ++++ pyproject.toml | 2 +- .../zb_find_email_format_response.py | 5 +- src/zerobouncesdk/zerobouncesdk.py | 283 +++++--- tests/test_offline_api.py | 621 ++++++++++++++++++ tests/zero_bounce_test_case.py | 21 +- 9 files changed, 1052 insertions(+), 87 deletions(-) create mode 100644 AGENTS.md create mode 100644 CHANGELOG.md create mode 100644 tests/test_offline_api.py diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 56343dc..286c8c0 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -86,9 +86,12 @@ jobs: - name: Install dependencies run: pip install -e ".[test]" + - name: Compile + run: python -m compileall -q src tests + - name: Run tests run: | - python -m pytest tests/zero_bounce_test_case.py -v \ + python -m pytest \ --cov=zerobouncesdk --cov-report=term-missing --cov-fail-under=50 publish: diff --git a/.github/workflows/sdk_ci.yml b/.github/workflows/sdk_ci.yml index e4d06ba..f0b2ae7 100644 --- a/.github/workflows/sdk_ci.yml +++ b/.github/workflows/sdk_ci.yml @@ -27,7 +27,10 @@ jobs: - name: Install dependencies run: pip install -e ".[test]" + - name: Compile + run: python -m compileall -q src tests + - name: Run unit tests with coverage run: | - python -m pytest tests/zero_bounce_test_case.py -v \ + python -m pytest \ --cov=zerobouncesdk --cov-report=term-missing --cov-fail-under=50 diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..847282d --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,57 @@ +# AGENTS.md + +Guidance for AI coding agents working in this repository. + +## Project + +`zero-bounce-python-sdk-setup` is the ZeroBounce Python SDK, published to PyPI as `zerobouncesdk`. Callers construct a `ZeroBounce` with an API key and call methods that return response objects or raise `ZBApiException` / `ZBClientException`. HTTP goes through the `requests` library inside `src/zerobouncesdk/zerobouncesdk.py`. + +## Layout + +- `src/zerobouncesdk/zerobouncesdk.py`: the client. `_get`, `_post`, `_post_form`, and `_request_json` are the HTTP entry points. +- `src/zerobouncesdk/zb_*.py`: response models, enums, and options. +- `tests/zero_bounce_test_case.py`: scripted unit tests. They patch `zerobouncesdk.zerobouncesdk.requests`. +- `tests/test_offline_api.py`: offline suite. It patches `requests.sessions.Session.send`, so the real `requests` client builds the URL and body and the fake answers without the network. +- `tests/zero_bounce_integration_test.py`: live API tests. `pytest` does not collect this file. Run it with `python -m unittest tests.zero_bounce_integration_test` when `ZEROBOUNCE_API_KEY` is set. +- `pyproject.toml`: the only version field, `version = "x.y.z"`. `setup.py` calls `setup()` and reads that file. + +## Commands + +```bash +pip install -e ".[test]" +python -m compileall -q src tests # syntax check; there is no separate typechecker or linter +python -m pytest # unit tests and tests/test_offline_api.py +``` + +CI (`.github/workflows/sdk_ci.yml`) compiles, then runs `python -m pytest` with coverage. There is no MyPy or Ruff config. Do not add one that fails the existing tree unless asked. + +## Conventions + +- **One `ZeroBounce` per API key.** The class is not a singleton. Do not reuse one instance across users or requests. The offline concurrency test constructs a separate client per key. +- **Column indexes are 1-based, and the SDK does not add 1.** Pass the API's index. `0` is a real value and is sent. `None` on an optional column (`first_name_column`, and the S3 email column) omits the field. `send_file`'s email column is required and is sent as given, including `0`. `domain_column` and the list-evaluator email column are required: `None` is rejected, `0` is sent. +- **Optional parameters are omitted when unset.** `timeout`, `activity_data`, `verify_plus`, and `allow_phase_2` are not sent until the caller sets them. Explicit `False` is sent as `false`. `ip_address=None` is omitted; `""` is sent. +- **Booleans on form and multipart bodies must be the strings `true` and `false`.** A Python `bool` passed to `requests` becomes `True` or `False`, and the API will not treat that as a boolean. `_form_data` converts them. JSON bodies (`validate_batch`) keep JSON booleans. +- **`allow_phase_2` is validation `send_file` and `validate_s3file` only.** Scoring `send_file` and `score_s3file` do not send it. `activity_data` on get-file is validation `get_file` only. +- **`validate_batch` uses `api.zerobounce.net`**, not `bulkapi.zerobounce.net`. It follows `base_url`. +- **Failures raise.** `ZBClientException` means the arguments were rejected and no request was sent. `ZBApiException` means the API failed. The message must be short: no raw HTML, no megabyte body, no `JSONDecodeError` / `Expecting value` text, and no `Error: Error:` prefix. An account-level `validate_batch` error (`email_address` `all`) raises. A per-address error stays on the result's `errors` list. +- **Do not follow redirects.** Every `requests` call passes `allow_redirects=False`. A 3xx response is a `ZBApiException`. A redirect must not be fetched, because the `Location` could be another host and the next request can carry the API key. +- **Do not commit `.env`.** It is gitignored and may hold a real key. Tests must not call the live API. +- Match the surrounding style. Do not reformat unrelated code. + +## Changelog + +`CHANGELOG.md` follows [Keep a Changelog 1.1.0](https://keepachangelog.com/en/1.1.0/), and the project adheres to [Semantic Versioning 2.0.0](https://semver.org/spec/v2.0.0.html). + +- Add every user-visible change to `## [Unreleased]` in the same commit as the change. +- Group entries under these headings, in this order, omitting empty ones: `Added`, `Changed`, `Deprecated`, `Removed`, `Fixed`, `Security`. +- Write for SDK users: name the method, option, or behaviour, and say what changed. Leave out CI-only changes, internal refactors, and fixes to code that was never released. +- At 2.y.z, a breaking change bumps the major version, a new method or option bumps the minor version, and fixes alone bump the patch version. + +## Releasing + +Publishing and pushing tags need a maintainer's explicit approval. Do not bump the version, tag, or publish unless they asked. + +1. Move `[Unreleased]` entries under `## [x.y.z] - YYYY-MM-DD`, leave an empty `[Unreleased]` section, and update the compare links at the bottom of `CHANGELOG.md`. +2. Set the same version in `pyproject.toml` (`version = "x.y.z"`). The publish workflow fails if it differs from the tag. That is the only version field. There is no runtime `__version__` and no user-agent constant. `setup.py` does not contain a version. +3. `python -m compileall -q src tests` and `python -m pytest`. +4. Tag `vx.y.z` on that commit and run the Publish workflow (`.github/workflows/publish.yml`) with that tag. It publishes `zerobouncesdk` to PyPI. diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..e89d96a --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,140 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [Unreleased] + +### Added + +- `validate` takes `timeout` (3–60 seconds), `activity_data`, `verify_plus`, and `use_post` (form-encoded `POST /validate`). +- `validate_batch` takes `timeout` (10–120 seconds), `activity_data`, and `verify_plus`. +- `ZBValidateResponse` reads `active_in_days` and `active_first_seen`. +- `score_email` (`GET /scoring`) and `score_email(..., use_post=True)` (form-encoded `POST /scoring`). +- Bulk email finder (`send_email_finder_file`, `email_finder_file_status`, `email_finder_get_file`, `email_finder_delete_file`, plus stream upload). +- Bulk domain search (`send_domain_search_file`, `domain_search_file_status`, `domain_search_get_file`, `domain_search_delete_file`, plus stream upload). +- List evaluator (`send_list_evaluator_file`, `get_list_evaluator_result`, `send_evaluated_file`). +- Filters (`add_filter`, `list_filters`, `delete_filter`). +- S3 bulk (`validate_s3file`, `score_s3file`, `export_s3file`, `list_s3files`). + +### Changed + +- A custom `base_url` must be an `https://` URL. +- API requests time out after 120 seconds unless the caller passes `timeout`. +- API requests do not follow redirects. + +### Fixed + +- A 403 HTML "Access denied" page (about 1.4 MB from the live API on `get_file`) is raised as a short `ZBApiException`. An HTML page with status 200 is an error as well, including `get_file`, `email_finder_get_file`, and `domain_search_get_file`, instead of a saved CSV. +- Empty and non-JSON bodies are a short `ZBApiException`. The message is no longer a chained `JSONDecodeError`, and a doubled `Error: Error:` / `[Error]: [Error]:` prefix is collapsed to one. +- `validate_batch` raises `ZBApiException` for an account-level failure (HTTP 200, error for address `all`, such as an invalid API key or no credits). An error for one address stays on the successful result. +- `send_file`, `scoring_send_file`, bulk email finder, domain search, and S3 uploads send `has_header_row`, `remove_duplicate`, and `remove_duplicates` as `true` or `false`. `has_header_row=False` is sent. +- `send_email_finder_file`, `send_domain_search_file`, and `send_list_evaluator_file` send column `0`. They previously treated `0` as empty and rejected it before the request. +- `find_email_format` raises `ZBApiException` when the API returns a `Message` error, the same way `find_domain` and `guess_format` already did. + +### Security + +- Error messages no longer include the raw HTML access-denied page. The live `get_file` page is about 1.4 MB and contains the API key. +- Redirects are not followed, so a response cannot send the client, or the API key, to another host. + +## [2.2.3] - 2026-06-18 + +### Fixed + +- `validate_batch` posts to `api.zerobounce.net/v2/validatebatch` instead of `bulkapi.zerobounce.net`. + +## [2.2.2] - 2026-06-10 + +### Added + +- `ZBValidateResponse.catchall_domain`, read from `catchall_domain`. + +## [2.2.0] - 2026-04-05 + +### Added + +- `send_file` / `send_file_stream` `allow_phase_2`, sent as `allow_phase_2` on validation bulk sendfile only. +- `ZBGetFileOptions` for `get_file` and `scoring_get_file`, with `download_type` (`phase_1`, `phase_2`, `combined`) and `activity_data` (validation `get_file` only). +- `ZBFileStatusResponse.file_phase_2_status`. + +### Fixed + +- `get_file` raises `ZBApiException` when the body is a JSON error, including HTTP 200 with `"success": false`, instead of saving that body as the results file. + +## [2.1.10] - 2026-03-03 + +### Changed + +- License the project under MIT. + +## [2.1.0] - 2026-03-03 + +### Added + +- `ZeroBounce(..., timeout=timedelta)` to set the HTTP timeout. +- `send_file_stream` and `scoring_send_file_stream`. + +### Fixed + +- `get_credits` and `validate_batch` tolerate a response that omits `Credits`, `email_batch`, or `errors`. + +## [2.0.1] - 2026-01-26 + +### Fixed + +- `ZBValidateSubStatus.gold` and unknown status values are recognized again. The 2.0.0 tag did not include them. + +## [2.0.0] - 2026-01-26 + +### Removed + +- `ZBValidateSubStatus.gold` and the fallback that kept an unknown status from raising. Restored in 2.0.1. + +## [1.1.3] - 2026-01-26 + +### Added + +- `find_email_format` and `find_domain`. +- `ZeroBounce(api_key, base_url)` with `ZBApiUrl` (`API_DEFAULT_URL`, `API_USA_URL`, `API_EU_URL`) or an https URL string. +- `ZBValidateSubStatus.gold` and `ZBValidateSubStatus.role_based_accept_all`. + +### Changed + +- An unknown status, sub-status, or confidence value warns and becomes `None` instead of raising. + +### Deprecated + +- `guess_format`. Use `find_email_format` or `find_domain`. + +## [1.1.2] - 2025-08-07 + +### Added + +- `ZBValidateSubStatus.accept_all`. + +## [1.1.1] - 2024-12-20 + +### Fixed + +- API error payloads raise `ZBApiException` with the error text. + +## [1.1.0] - 2023-09-12 + +### Added + +- Initial release: `get_credits`, `get_api_usage`, `get_activity`, `validate`, `validate_batch`, bulk and AI-scoring file methods, and `guess_format`. + +[Unreleased]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v2.2.4...HEAD +[2.2.3]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v2.2.2...v2.2.3 +[2.2.2]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v2.2.0...v2.2.2 +[2.2.0]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v2.1.10...v2.2.0 +[2.1.10]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v2.1.0...v2.1.10 +[2.1.0]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v2.0.1...v2.1.0 +[2.0.1]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v2.0.0...v2.0.1 +[2.0.0]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v1.1.3...v2.0.0 +[1.1.3]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v1.1.2...v1.1.3 +[1.1.2]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v1.1.1...v1.1.2 +[1.1.1]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/compare/v1.1.0...v1.1.1 +[1.1.0]: https://github.com/zerobounce/zero-bounce-python-sdk-setup/releases/tag/v1.1.0 diff --git a/pyproject.toml b/pyproject.toml index fbb7a4b..27594a6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,7 @@ Homepage = "https://github.com/zerobounce/zero-bounce-python-sdk-setup" [tool.pytest.ini_options] testpaths = ["tests"] -python_files = ["test_*.py", "*_test.py", "zero_bounce_test_case.py"] +python_files = ["test_*.py", "zero_bounce_test_case.py"] python_functions = ["test_*"] addopts = "-v --tb=short" filterwarnings = ["ignore::DeprecationWarning"] diff --git a/src/zerobouncesdk/zb_find_email_format_response.py b/src/zerobouncesdk/zb_find_email_format_response.py index 7108b3c..811262d 100644 --- a/src/zerobouncesdk/zb_find_email_format_response.py +++ b/src/zerobouncesdk/zb_find_email_format_response.py @@ -1,4 +1,4 @@ -from . import ZBConfidence +from . import ZBApiException, ZBConfidence from ._zb_response import ZBResponse @@ -25,6 +25,9 @@ class ZBFindEmailFormatResponse(ZBResponse): def __init__(self, data): super().__init__(data) + if "Message" in data or "message" in data: + message = data.get("message", data.get("Message")) + raise ZBApiException(message) if self.email_confidence and isinstance(self.email_confidence, str): self.email_confidence = ZBConfidence(self.email_confidence.lower()) diff --git a/src/zerobouncesdk/zerobouncesdk.py b/src/zerobouncesdk/zerobouncesdk.py index 0e15d16..c0283d0 100644 --- a/src/zerobouncesdk/zerobouncesdk.py +++ b/src/zerobouncesdk/zerobouncesdk.py @@ -1,4 +1,5 @@ import json +import re import warnings from datetime import date, timedelta import os @@ -30,6 +31,120 @@ ) +def _looks_like_html(body: str, content_type: str = "") -> bool: + if content_type and "text/html" in content_type.lower(): + return True + sample = (body or "").lstrip()[:2000].lower() + return ( + " str: + snippet = body[:8000] + lower = snippet.lower() + start = lower.find("", start) + if open_end < 0: + return "" + close = lower.find("", open_end) + if close < 0: + return "" + return snippet[open_end + 1:close].strip() + + +def _strip_error_prefix(message: str) -> str: + """Collapse a repeated ``Error:`` / ``[Error]:`` prefix and leave a single one.""" + stripped = message.strip() + changed = True + while changed: + changed = False + match = re.match(r"(?i)(?:error:\s*){2,}", stripped) + if match: + stripped = ("Error: " + stripped[match.end():]).strip() + changed = True + match = re.match(r"(?i)(?:\[error\]:\s*){2,}", stripped) + if match: + stripped = ("[Error]: " + stripped[match.end():]).strip() + changed = True + return stripped + + +def _message_from_body(body: str, status_code: int) -> str: + """Turn an HTTP body into a short message. Never return raw HTML or a parser exception.""" + trimmed = (body or "").strip() + if not trimmed: + return "Empty response from ZeroBounce (HTTP %s)" % status_code + if _looks_like_html(body): + title = _html_title(body) + if title and "access denied" in title.lower(): + return "Access denied" + if title: + return 'ZeroBounce returned an HTML error page ("%s") instead of an API response.' % title + return "ZeroBounce returned an HTML error page instead of an API response." + if trimmed[0] in "{[": + try: + parsed = json.loads(trimmed) + except ValueError: + return "ZeroBounce returned a response that is not valid JSON." + if isinstance(parsed, dict): + title = parsed.get("title") + if isinstance(title, str) and ("Access denied" in title or "1020" in title): + has_api_message = False + for key in ("error", "message", "error_message"): + value = parsed.get(key) + if isinstance(value, str) and value.strip(): + has_api_message = True + elif isinstance(value, list) and value: + has_api_message = True + if parsed.get("cloudflare_error") is True or not has_api_message: + return title + for key in ("message", "error", "error_message"): + value = parsed.get(key) + if isinstance(value, str) and value.strip(): + return _strip_error_prefix(value) + if isinstance(value, list) and value and isinstance(value[0], str) and value[0].strip(): + return _strip_error_prefix(value[0]) + if len(trimmed) > 500: + return "ZeroBounce returned an error (HTTP %s)." % status_code + return trimmed + if len(trimmed) > 1000 or any(char in trimmed for char in "{}[]<>"): + return "ZeroBounce returned a response that is not valid JSON." + message = _strip_error_prefix(trimmed) + if not message: + return "Empty response from ZeroBounce (HTTP %s)" % status_code + return message + + +def _form_data(data): + """Encode booleans as the strings ``true`` and ``false`` for form and multipart bodies. + + ``requests`` would otherwise send a Python bool as ``True`` or ``False``. + """ + if not isinstance(data, dict): + return data + encoded = {} + for key, value in data.items(): + if isinstance(value, bool): + encoded[key] = "true" if value else "false" + else: + encoded[key] = value + return encoded + + +def _response_text(response) -> str: + content = getattr(response, "content", None) + if isinstance(content, (bytes, bytearray)): + return content.decode("utf-8", errors="replace") + text = getattr(response, "text", "") + return text if isinstance(text, str) else "" + + class ZeroBounce: """The ZeroBounce main class. All the requests are implemented here.""" @@ -109,24 +224,6 @@ def get_file_json_indicates_error(body: str) -> bool: return True return "success" in o - @staticmethod - def _format_get_file_error_message(body: str) -> str: - try: - o = json.loads(body) - except ValueError: - return body if body else "Invalid getfile response" - if not isinstance(o, dict): - return body - for k in ("message", "error", "error_message"): - v = o.get(k) - if not v: - continue - if isinstance(v, str) and v.strip(): - return v.strip() - if isinstance(v, list) and v and isinstance(v[0], str): - return v[0].strip() - return body - @staticmethod def _should_treat_get_file_body_as_error(body: str, content_type: str) -> bool: ct = content_type.lower() @@ -141,33 +238,84 @@ def _redact(self, text) -> str: text = text.replace(secret, "***") return text + def _read_json_payload(self, response): + body = _response_text(response) + status = getattr(response, "status_code", 0) or 0 + headers = getattr(response, "headers", None) or {} + content_type = headers.get("Content-Type") or headers.get("content-type") or "" + if 300 <= status < 400: + raise ZBApiException( + self._redact("ZeroBounce returned a redirect, which this client does not follow.") + ) + if status > 299 or _looks_like_html(body, content_type) or not body.strip(): + raise ZBApiException(self._redact(_message_from_body(body, status))) + try: + payload = response.json() + except ValueError: + raise ZBApiException(self._redact(_message_from_body(body, status))) from None + if isinstance(payload, dict): + title = payload.get("title") + if isinstance(title, str) and payload.get("cloudflare_error") is True: + raise ZBApiException(self._redact(title)) + error = None + for key in ("error", "Error"): + if key in payload and payload[key]: + error = payload[key] + break + if isinstance(error, str) and error.strip(): + raise ZBApiException(self._redact(_strip_error_prefix(error))) + if error: + raise ZBApiException(self._redact(str(error))) + success = payload.get("success") + failed = success is False or ( + isinstance(success, str) and success.lower() == "false" + ) + if failed: + message = payload.get("message") or payload.get("error_message") or "Request failed" + if isinstance(message, list): + message = message[0] if message else "Request failed" + text = str(message).strip() or "Request failed" + raise ZBApiException(self._redact(_strip_error_prefix(text))) + return payload + + def _raise_if_download_error(self, response, body_str: str, content_type: str) -> None: + status = response.status_code + if 300 <= status < 400: + raise ZBApiException( + self._redact("ZeroBounce returned a redirect, which this client does not follow.") + ) + if ( + status > 299 + or _looks_like_html(body_str, content_type) + or self._should_treat_get_file_body_as_error(body_str, content_type) + or not body_str.strip() + or "text/plain" in content_type.lower() + ): + raise ZBApiException(self._redact(_message_from_body(body_str, status))) + def _get(self, url, response_class, params=None): if not params: params = {} params["api_key"] = self._api_key try: - response = requests.get(url, params=params, timeout=self._timeout_s) + response = requests.get(url, params=params, timeout=self._timeout_s, allow_redirects=False) except RequestException as e: raise type(e)(self._redact(e)) from None - - try: - json_response = response.json() - except ValueError as e: - raise ZBApiException from e - - error = json_response.pop("error", None) - if error: - raise ZBApiException(error) - return response_class(json_response) + return response_class(self._read_json_payload(response)) def _post(self, url, response_class, data=None, json=None, files=None): - response = requests.post(url, data=data, json=json, files=files, timeout=self._timeout_s) try: - json_response = response.json() - except ValueError as e: - raise ZBApiException('Request not processed succesfully. Status code %s' % response.status_code) - - return response_class(json_response) + response = requests.post( + url, + data=_form_data(data), + json=json, + files=files, + timeout=self._timeout_s, + allow_redirects=False, + ) + except RequestException as e: + raise type(e)(self._redact(e)) from None + return response_class(self._read_json_payload(response)) def _post_form(self, url, response_class, data=None): if not data: @@ -175,19 +323,12 @@ def _post_form(self, url, response_class, data=None): data = dict(data) data["api_key"] = self._api_key try: - response = requests.post(url, data=data, timeout=self._timeout_s) + response = requests.post( + url, data=_form_data(data), timeout=self._timeout_s, allow_redirects=False + ) except RequestException as e: raise type(e)(self._redact(e)) from None - - try: - json_response = response.json() - except ValueError as e: - raise ZBApiException from e - - error = json_response.pop("error", None) - if error: - raise ZBApiException(error) - return response_class(json_response) + return response_class(self._read_json_payload(response)) def _request_json(self, method: str, url: str, *, params=None, data=None, files=None): if params is None: @@ -200,18 +341,14 @@ def _request_json(self, method: str, url: str, *, params=None, data=None, files= method, url, params=params, - data=data, + data=_form_data(data), files=files, timeout=self._timeout_s, + allow_redirects=False, ) except RequestException as e: raise type(e)(self._redact(e)) from None - try: - return response.json() - except ValueError as e: - raise ZBApiException( - f"Request not processed successfully. Status code {response.status_code}" - ) from e + return self._read_json_payload(response) def _send_bulk_root_multipart( self, url: str, file_path: str, fields: dict @@ -253,18 +390,14 @@ def _bulk_root_get_file( f"{base_url}/getfile", params=params, timeout=self._timeout_s, + allow_redirects=False, ) except RequestException as e: raise type(e)(self._redact(e)) from None body_bytes = response.content or b"" body_str = body_bytes.decode("utf-8", errors="replace") content_type = response.headers.get("Content-Type") or "" - if response.status_code > 299: - if body_str.lstrip().startswith("{"): - raise ZBApiException(self._redact(self._format_get_file_error_message(body_str))) - raise ZBApiException(self._redact(body_str or f"HTTP {response.status_code}")) - if self._should_treat_get_file_body_as_error(body_str, content_type): - raise ZBApiException(self._redact(self._format_get_file_error_message(body_str))) + self._raise_if_download_error(response, body_str, content_type) dirname = os.path.dirname(download_path) if dirname: os.makedirs(dirname, exist_ok=True) @@ -505,10 +638,18 @@ def validate_batch( payload["activity_data"] = activity_data if verify_plus is not None: payload["verify_plus"] = verify_plus - return self._post( + result = self._post( f"{self._base_url}/validatebatch", ZBValidateBatchResponse, json=payload) + for batch_error in result.errors or []: + address = (batch_error.email_address or "").strip().lower() + if address == "all": + text = batch_error.error or "Invalid API Key or your account ran out of credits" + if isinstance(text, str): + text = _strip_error_prefix(text) + raise ZBApiException(text) + return result def _send_file( self, @@ -840,20 +981,14 @@ def _get_file( f"{self.SCORING_BASE_URL if scoring else self.BULK_BASE_URL}/getfile", params=params, timeout=self._timeout_s, + allow_redirects=False, ) except RequestException as e: raise type(e)(self._redact(e)) from None body_bytes = response.content or b"" body_str = body_bytes.decode("utf-8", errors="replace") content_type = response.headers.get("Content-Type") or "" - - if response.status_code > 299: - if body_str.lstrip().startswith("{"): - raise ZBApiException(self._redact(self._format_get_file_error_message(body_str))) - raise ZBApiException(self._redact(body_str or f"HTTP {response.status_code}")) - - if self._should_treat_get_file_body_as_error(body_str, content_type): - raise ZBApiException(self._redact(self._format_get_file_error_message(body_str))) + self._raise_if_download_error(response, body_str, content_type) dirname = os.path.dirname(download_path) if dirname: @@ -988,7 +1123,7 @@ def send_email_finder_file( full_name_column: int = None, has_header_row: bool = False, ) -> ZBSendFileResponse: - if not domain_column: + if domain_column is None: raise ZBClientException("Empty parameter: domain_column") fields = {"domain_column": domain_column, "has_header_row": has_header_row} if first_name_column is not None: @@ -1014,7 +1149,7 @@ def send_email_finder_file_stream( full_name_column: int = None, has_header_row: bool = False, ) -> ZBSendFileResponse: - if not domain_column: + if domain_column is None: raise ZBClientException("Empty parameter: domain_column") fields = {"domain_column": domain_column, "has_header_row": has_header_row} if first_name_column is not None: @@ -1049,7 +1184,7 @@ def send_domain_search_file( domain_column: int, has_header_row: bool = False, ) -> ZBSendFileResponse: - if not domain_column: + if domain_column is None: raise ZBClientException("Empty parameter: domain_column") return self._send_bulk_root_multipart( f"{self.DOMAIN_SEARCH_BASE_URL}/sendfile", @@ -1064,7 +1199,7 @@ def send_domain_search_file_stream( domain_column: int, has_header_row: bool = False, ) -> ZBSendFileResponse: - if not domain_column: + if domain_column is None: raise ZBClientException("Empty parameter: domain_column") return self._send_bulk_root_multipart_stream( f"{self.DOMAIN_SEARCH_BASE_URL}/sendfile", @@ -1087,7 +1222,7 @@ def domain_search_delete_file(self, file_id: str) -> ZBDeleteFileResponse: def send_list_evaluator_file( self, file_path: str, email_address_column: int ) -> ZBSendFileResponse: - if not email_address_column: + if email_address_column is None: raise ZBClientException("Empty parameter: email_address_column") data = { "api_key": self._api_key, @@ -1108,7 +1243,7 @@ def send_list_evaluator_file_stream( file_name: str, email_address_column: int, ) -> ZBSendFileResponse: - if not email_address_column: + if email_address_column is None: raise ZBClientException("Empty parameter: email_address_column") data = { "api_key": self._api_key, diff --git a/tests/test_offline_api.py b/tests/test_offline_api.py new file mode 100644 index 0000000..bb8283b --- /dev/null +++ b/tests/test_offline_api.py @@ -0,0 +1,621 @@ +"""Offline API suite. + +A fake ``requests`` transport records every call and answers without the network. +Error modes are selected by the API key prefix and follow the live API. +""" + +import json +import random +import tempfile +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from datetime import date +from email.parser import BytesParser +from email.policy import HTTP +from io import BytesIO +from pathlib import Path +from unittest import TestCase +from unittest.mock import patch +from urllib.parse import parse_qs, urlparse + +import requests +from requests import Response + +from zerobouncesdk import ( + ZBApiException, + ZBClientException, + ZBGetFileOptions, + ZBValidateBatchElement, + ZeroBounce, +) + +_HTML_PAGE = ( + "Access denied | zerobounce.net" + + ("Access denied. " * 4000) + + "" +) + + +class RecordedRequest: + def __init__(self, method, url, body, headers): + parsed = urlparse(url) + self.method = method.upper() + self.url = url + self.host = parsed.hostname + self.path = parsed.path + self.query = {k: v[-1] for k, v in parse_qs(parsed.query, keep_blank_values=True).items()} + self.headers = {k.lower(): v for k, v in (headers or {}).items()} + self.body = body if isinstance(body, (bytes, bytearray)) else (body or "").encode() + content_type = self.headers.get("content-type", "") + self.json = None + self.form = {} + self.files = {} + if "application/json" in content_type and self.body: + self.json = json.loads(self.body.decode()) + elif "multipart/form-data" in content_type and self.body: + self.form, self.files = _parse_multipart(content_type, self.body) + elif self.body and "application/x-www-form-urlencoded" in content_type: + self.form = { + k: v[-1] + for k, v in parse_qs(self.body.decode(), keep_blank_values=True).items() + } + elif self.body and self.body[:1] in (b"{", b"["): + self.json = json.loads(self.body.decode()) + + @property + def api_key(self): + if "api_key" in self.query: + return self.query["api_key"] + if self.json and isinstance(self.json, dict) and "api_key" in self.json: + return self.json["api_key"] + return self.form.get("api_key") + + def field(self, name): + if name in self.query: + return self.query[name] + if self.json and isinstance(self.json, dict) and name in self.json: + return self.json[name] + if name in self.form: + return self.form[name] + return None + + def has_field(self, name): + if name in self.query: + return True + if self.json and isinstance(self.json, dict) and name in self.json: + return True + return name in self.form + + +def _parse_multipart(content_type, body): + raw = b"Content-Type: " + content_type.encode() + b"\r\n\r\n" + body + message = BytesParser(policy=HTTP).parsebytes(raw) + form = {} + files = {} + for part in message.iter_parts(): + name = part.get_param("name", header="content-disposition") + filename = part.get_filename() + payload = part.get_payload(decode=True) or b"" + if filename: + files[name] = {"filename": filename, "data": payload} + else: + form[name] = payload.decode() + return form, files + + +class FakeTransport: + def __init__(self): + self.requests = [] + self._lock = threading.Lock() + + def send(self, session, request, **kwargs): + time.sleep(random.uniform(0.005, 0.02)) + recorded = RecordedRequest(request.method, request.url, request.body, request.headers) + with self._lock: + self.requests.append(recorded) + return self._reply(recorded) + + def _reply(self, recorded): + key = recorded.api_key or "" + if key.startswith("redirect-"): + return _response(302, b"", "text/plain", headers={"Location": "https://evil.example/steal"}) + if key.startswith("denied-"): + return _response(403, _HTML_PAGE.encode(), "text/html; charset=UTF-8") + if key.startswith("html200-"): + page = "Not a filenot csv" + return _response(200, page.encode(), "text/html") + if key.startswith("empty-"): + return _response(200, b"", "text/plain") + if key.startswith("malformed-"): + return _response(200, b"this is not json {{{", "application/json") + if key.startswith("invalid-") or key.startswith("nocredits-"): + return self._account_error(recorded, key) + if key.startswith("itemerr-") and recorded.path.endswith("/validatebatch"): + payload = { + "email_batch": [{"address": "good@example.com", "status": "valid"}], + "errors": [{"email_address": "bad@example.com", "error": "Invalid email address"}], + } + return _response(200, json.dumps(payload).encode(), "application/json") + if key.startswith("doubled-"): + return _response( + 200, + json.dumps({"error": "Error: Error: Invalid API key"}).encode(), + "application/json", + ) + return _response(200, json.dumps(self._success(recorded, key)).encode(), "application/json") + + def _account_error(self, recorded, key): + no_credits = key.startswith("nocredits-") + text = "Your account ran out of credits" if no_credits else "Invalid API key or your account ran out of credits" + path = recorded.path + if path.endswith("/getcredits"): + return _response(200, b'{"Credits":"-1"}', "application/json") + if path.endswith("/getapiusage") and not no_credits: + return _response(200, b'{"error":"Invalid API key"}', "application/json") + if path.endswith("/validatebatch"): + payload = { + "email_batch": [], + "errors": [{"email_address": "all", "error": "Invalid API Key or your account ran out of credits"}], + } + return _response(200, json.dumps(payload).encode(), "application/json") + if path.endswith("/getfile"): + return _response(403, _HTML_PAGE.encode(), "text/html; charset=UTF-8") + if path.endswith("/deletefile") or path.endswith("/filestatus") and "email-finder" in path or path.endswith("/filestatus") and "domain-search" in path: + return _response(403, _HTML_PAGE.encode(), "text/html; charset=UTF-8") + if path.endswith("/filestatus"): + payload = {"success": "False", "message": ["api_key is invalid"]} + return _response(401, json.dumps(payload).encode(), "application/json") + if path.endswith("/sendfile") or path.rstrip("/").endswith("listevaluator") or "send-evaluated-file" in path: + payload = {"success": "False", "message": ["api_key is invalid"]} + return _response(401, json.dumps(payload).encode(), "application/json") + if path.endswith("/listS3files") or "S3file" in path or path.rstrip("/").endswith("listevaluator") or "/listevaluator/" in path: + payload = {"success": "False", "message": ["api_key is invalid"]} + status = 401 + return _response(status, json.dumps(payload).encode(), "application/json") + if "/filters/" in path: + payload = {"Error": text} + return _response(200, json.dumps(payload).encode(), "application/json") + if path.endswith("/guessformat"): + return _response(200, json.dumps({"Message": text}).encode(), "application/json") + return _response(200, json.dumps({"error": text}).encode(), "application/json") + + def _success(self, recorded, key): + path = recorded.path + echo = {"echo_key": key} + if path.endswith("/getfile"): + # Replaced below with CSV bytes. Marker keeps the branch obvious. + return {"__csv__": key} + if path.endswith("/getcredits"): + return {"Credits": key} + if path.endswith("/getapiusage"): + return {"total": 1, "start_date": "1/1/2024", "end_date": "1/2/2024", **echo} + if path.endswith("/activity"): + return {"found": True, "active_in_days": "30", **echo} + if path.endswith("/scoring"): + return {"email": recorded.field("email"), "score": 9, **echo} + if path.endswith("/validate"): + return {"address": recorded.field("email"), "status": "valid", **echo} + if path.endswith("/validatebatch"): + return { + "email_batch": [{"address": "a@b.com", "status": "valid", "echo_key": key}], + "errors": [], + } + if path.endswith("/filestatus"): + return { + "success": True, + "file_id": recorded.field("file_id"), + "file_name": "emails.csv", + "upload_date": "2023-03-28T12:30:18Z", + "file_status": "Complete", + "complete_percentage": "100%", + **echo, + } + if path.endswith("/deletefile"): + return {"success": True, "message": "File Deleted", "file_id": recorded.field("file_id"), "file_name": "emails.csv", **echo} + if path.endswith("/sendfile") or "listevaluator" in path or "send-evaluated-file" in path or path.endswith("S3file"): + return {"success": True, "message": "File Accepted", "file_id": "file-1", "file_name": "emails.csv", **echo} + if path.endswith("/guessformat"): + return { + "email": "ada@example.com", + "domain": recorded.field("domain") or "example.com", + "company_name": recorded.field("company_name") or "", + "format": "first", + "status": "valid", + "sub_status": "", + "confidence": "high", + "email_confidence": "high", + "did_you_mean": "", + "failure_reason": "", + "other_domain_formats": [], + **echo, + } + if "/filters/" in path: + return {"Message": "Filter successfully added", **echo} + if path.endswith("/listS3files"): + return [{"name": "uploads/a.csv", "echo_key": key}] + if "/listevaluator/" in path: + return {"file_status": "Complete", **echo} + return {"ok": True, **echo} + + def __call__(self, request, **kwargs): + reply = self.send(None, request, **kwargs) + # getfile success is CSV, not the JSON marker. + if reply.headers.get("Content-Type", "").startswith("application/json"): + try: + payload = json.loads(reply.content.decode()) + except ValueError: + return reply + if isinstance(payload, dict) and "__csv__" in payload: + key = payload["__csv__"] + return _response(200, f"email\n{key}\n".encode(), "text/csv") + return reply + + +def _response(status, body, content_type, headers=None): + response = Response() + response.status_code = status + response._content = body + response.headers["Content-Type"] = content_type + response.encoding = "utf-8" + response.url = "https://api.zerobounce.net/v2/unused" + if headers: + for name, value in headers.items(): + response.headers[name] = value + return response + + +class OfflineApiTest(TestCase): + def setUp(self): + self.transport = FakeTransport() + + def send(session, request, **kwargs): + return self.transport(request, **kwargs) + + self._patch = patch("requests.sessions.Session.send", send) + self._patch.start() + self.addCleanup(self._patch.stop) + self.client = ZeroBounce("client-key") + + def _only(self): + self.assertEqual(len(self.transport.requests), 1) + return self.transport.requests[0] + + def _csv(self, name="rows.csv"): + path = Path(tempfile.mkdtemp()) / name + path.write_text("email,name\nada@example.com,Ada\n") + return path + + def test_every_public_method_hits_the_right_endpoint(self): + csv = self._csv() + stream = BytesIO(b"email\nada@example.com\n") + out = Path(tempfile.mkdtemp()) / "out.csv" + cases = [ + ("get_credits", lambda: self.client.get_credits(), "GET", "api.zerobounce.net", "/v2/getcredits"), + ("get_api_usage", lambda: self.client.get_api_usage(date(2024, 1, 2), date(2024, 1, 3)), "GET", "api.zerobounce.net", "/v2/getapiusage"), + ("get_activity", lambda: self.client.get_activity("ada@example.com"), "GET", "api.zerobounce.net", "/v2/activity"), + ("score_email", lambda: self.client.score_email("ada@example.com"), "GET", "api.zerobounce.net", "/v2/scoring"), + ("score_email_post", lambda: self.client.score_email("ada@example.com", use_post=True), "POST", "api.zerobounce.net", "/v2/scoring"), + ("validate", lambda: self.client.validate("ada@example.com"), "GET", "api.zerobounce.net", "/v2/validate"), + ("validate_post", lambda: self.client.validate("ada@example.com", use_post=True), "POST", "api.zerobounce.net", "/v2/validate"), + ("validate_batch", lambda: self.client.validate_batch([ZBValidateBatchElement("ada@example.com", "1.2.3.4")]), "POST", "api.zerobounce.net", "/v2/validatebatch"), + ("send_file", lambda: self.client.send_file(str(csv), 1), "POST", "bulkapi.zerobounce.net", "/v2/sendfile"), + ("send_file_stream", lambda: self.client.send_file_stream(BytesIO(b"a@b.com\n"), "a.csv", 1), "POST", "bulkapi.zerobounce.net", "/v2/sendfile"), + ("scoring_send_file", lambda: self.client.scoring_send_file(str(csv), 1), "POST", "bulkapi.zerobounce.net", "/v2/scoring/sendfile"), + ("scoring_send_file_stream", lambda: self.client.scoring_send_file_stream(stream, "a.csv", 1), "POST", "bulkapi.zerobounce.net", "/v2/scoring/sendfile"), + ("file_status", lambda: self.client.file_status("file-9"), "GET", "bulkapi.zerobounce.net", "/v2/filestatus"), + ("scoring_file_status", lambda: self.client.scoring_file_status("file-9"), "GET", "bulkapi.zerobounce.net", "/v2/scoring/filestatus"), + ("get_file", lambda: self.client.get_file("file-9", str(out)), "GET", "bulkapi.zerobounce.net", "/v2/getfile"), + ("scoring_get_file", lambda: self.client.scoring_get_file("file-9", str(out) + ".score"), "GET", "bulkapi.zerobounce.net", "/v2/scoring/getfile"), + ("delete_file", lambda: self.client.delete_file("file-9"), "GET", "bulkapi.zerobounce.net", "/v2/deletefile"), + ("scoring_delete_file", lambda: self.client.scoring_delete_file("file-9"), "GET", "bulkapi.zerobounce.net", "/v2/scoring/deletefile"), + ("send_email_finder_file", lambda: self.client.send_email_finder_file(str(csv), 1), "POST", "bulkapi.zerobounce.net", "/email-finder/sendfile"), + ("send_email_finder_file_stream", lambda: self.client.send_email_finder_file_stream(BytesIO(b"example.com\n"), "f.csv", 1), "POST", "bulkapi.zerobounce.net", "/email-finder/sendfile"), + ("email_finder_file_status", lambda: self.client.email_finder_file_status("file-9"), "GET", "bulkapi.zerobounce.net", "/email-finder/filestatus"), + ("email_finder_get_file", lambda: self.client.email_finder_get_file("file-9", str(out) + ".finder"), "GET", "bulkapi.zerobounce.net", "/email-finder/getfile"), + ("email_finder_delete_file", lambda: self.client.email_finder_delete_file("file-9"), "GET", "bulkapi.zerobounce.net", "/email-finder/deletefile"), + ("send_domain_search_file", lambda: self.client.send_domain_search_file(str(csv), 1), "POST", "bulkapi.zerobounce.net", "/domain-search/sendfile"), + ("send_domain_search_file_stream", lambda: self.client.send_domain_search_file_stream(BytesIO(b"example.com\n"), "d.csv", 1), "POST", "bulkapi.zerobounce.net", "/domain-search/sendfile"), + ("domain_search_file_status", lambda: self.client.domain_search_file_status("file-9"), "GET", "bulkapi.zerobounce.net", "/domain-search/filestatus"), + ("domain_search_get_file", lambda: self.client.domain_search_get_file("file-9", str(out) + ".domain"), "GET", "bulkapi.zerobounce.net", "/domain-search/getfile"), + ("domain_search_delete_file", lambda: self.client.domain_search_delete_file("file-9"), "GET", "bulkapi.zerobounce.net", "/domain-search/deletefile"), + ("send_list_evaluator_file", lambda: self.client.send_list_evaluator_file(str(csv), 1), "POST", "bulkapi.zerobounce.net", "/v2/listevaluator/"), + ("send_list_evaluator_file_stream", lambda: self.client.send_list_evaluator_file_stream(BytesIO(b"a@b.com\n"), "l.csv", 1), "POST", "bulkapi.zerobounce.net", "/v2/listevaluator/"), + ("get_list_evaluator_result", lambda: self.client.get_list_evaluator_result("file-9"), "GET", "bulkapi.zerobounce.net", "/v2/listevaluator/file-9/"), + ("send_evaluated_file", lambda: self.client.send_evaluated_file(str(csv)), "POST", "bulkapi.zerobounce.net", "/v2/send-evaluated-file/"), + ("add_filter", lambda: self.client.add_filter("allow", "domain", "example.com"), "POST", "api.zerobounce.net", "/v2/filters/add"), + ("list_filters", lambda: self.client.list_filters(), "GET", "api.zerobounce.net", "/v2/filters/list"), + ("list_filters_post", lambda: self.client.list_filters(use_post=True), "POST", "api.zerobounce.net", "/v2/filters/list"), + ("delete_filter", lambda: self.client.delete_filter("allow", "domain", "example.com"), "POST", "api.zerobounce.net", "/v2/filters/delete"), + ("list_s3files", lambda: self.client.list_s3files(bucket="b", search="a"), "GET", "bulkapi.zerobounce.net", "/listS3files"), + ("validate_s3file", lambda: self.client.validate_s3file("lists/jan.csv"), "POST", "bulkapi.zerobounce.net", "/validateS3file"), + ("score_s3file", lambda: self.client.score_s3file("lists/jan.csv"), "POST", "bulkapi.zerobounce.net", "/scoreS3file"), + ("export_s3file", lambda: self.client.export_s3file("file-9", "export-bucket"), "POST", "bulkapi.zerobounce.net", "/exportS3file"), + ("guess_format", lambda: self.client.guess_format("example.com", first_name="Ada"), "GET", "api.zerobounce.net", "/v2/guessformat"), + ("find_email_format", lambda: self.client.find_email_format("Ada", domain="example.com"), "GET", "api.zerobounce.net", "/v2/guessformat"), + ("find_domain", lambda: self.client.find_domain(domain="example.com"), "GET", "api.zerobounce.net", "/v2/guessformat"), + ] + for name, call, method, host, path in cases: + self.transport.requests.clear() + call() + recorded = self._only() + self.assertEqual(recorded.method, method, name) + self.assertEqual(recorded.host, host, name) + self.assertEqual(recorded.path, path, name) + self.assertEqual(recorded.api_key, "client-key", name) + + def test_validate_batch_posts_json_to_the_api_host(self): + self.client.validate_batch( + [ZBValidateBatchElement("ada@example.com", "0.0.0.0")], + timeout=30, + activity_data=False, + verify_plus=True, + ) + recorded = self._only() + self.assertNotIn("bulkapi", recorded.host) + batch = recorded.json["email_batch"] + self.assertEqual(batch[0]["email_address"], "ada@example.com") + self.assertEqual(batch[0]["ip_address"], "0.0.0.0") + self.assertEqual(recorded.json["timeout"], 30) + self.assertIs(recorded.json["activity_data"], False) + self.assertIs(recorded.json["verify_plus"], True) + + def test_optional_parameters_are_omitted_until_set(self): + self.client.validate("ada@example.com") + recorded = self._only() + self.assertEqual(recorded.query["email"], "ada@example.com") + for name in ("ip_address", "timeout", "activity_data", "verify_plus"): + self.assertNotIn(name, recorded.query) + self.transport.requests.clear() + self.client.validate( + "ada@example.com", + ip_address="203.0.113.8", + timeout=10, + activity_data=False, + verify_plus=False, + ) + recorded = self._only() + self.assertEqual(recorded.query["ip_address"], "203.0.113.8") + self.assertEqual(recorded.query["timeout"], "10") + self.assertEqual(recorded.query["activity_data"], "false") + self.assertEqual(recorded.query["verify_plus"], "false") + + def test_query_values_are_percent_encoded(self): + client = ZeroBounce("key with space&plus") + client.validate("a b@ex.com") + recorded = self._only() + self.assertNotIn(" ", recorded.url) + self.assertIn("key+with+space%26plus", recorded.url) + self.assertIn("a+b%40ex.com", recorded.url) + self.assertEqual(recorded.api_key, "key with space&plus") + self.assertEqual(recorded.query["email"], "a b@ex.com") + + def test_column_zero_is_sent_and_unset_columns_are_left_out(self): + csv = self._csv() + self.client.send_file(str(csv), 0, first_name_column=0, has_header_row=False, remove_duplicate=False) + recorded = self._only() + self.assertEqual(recorded.form["email_address_column"], "0") + self.assertEqual(recorded.form["first_name_column"], "0") + self.assertNotIn("last_name_column", recorded.form) + self.assertNotIn("gender_column", recorded.form) + self.assertNotIn("ip_address_column", recorded.form) + self.assertNotIn("allow_phase_2", recorded.form) + self.assertEqual(recorded.form["has_header_row"], "false") + self.assertEqual(recorded.form["remove_duplicate"], "false") + + self.transport.requests.clear() + self.client.send_email_finder_file(str(csv), 0, has_header_row=False) + recorded = self._only() + self.assertEqual(recorded.form["domain_column"], "0") + self.assertEqual(recorded.form["has_header_row"], "false") + self.assertNotIn("first_name_column", recorded.form) + + self.transport.requests.clear() + self.client.send_domain_search_file(str(csv), 0, has_header_row=False) + self.assertEqual(self._only().form["has_header_row"], "false") + + self.transport.requests.clear() + self.client.send_list_evaluator_file(str(csv), 0) + self.assertEqual(self._only().form["email_address_column"], "0") + + self.transport.requests.clear() + self.client.validate_s3file("a.csv", email_address_column=0, has_header_row=False, remove_duplicate=False, allow_phase_2=False) + recorded = self._only() + self.assertEqual(recorded.form["email_address_column"], "0") + self.assertEqual(recorded.form["has_header_row"], "false") + self.assertEqual(recorded.form["remove_duplicate"], "false") + self.assertEqual(recorded.form["allow_phase_2"], "false") + self.assertNotIn("remove_duplicates", recorded.form) + + self.transport.requests.clear() + self.client.scoring_send_file(str(csv), 1, has_header_row=True, remove_duplicate=True) + recorded = self._only() + self.assertEqual(recorded.form["has_header_row"], "true") + self.assertEqual(recorded.form["remove_duplicate"], "true") + self.assertNotIn("allow_phase_2", recorded.form) + + def test_get_file_options_and_scoring_omits_activity_data(self): + out = Path(tempfile.mkdtemp()) / "out.csv" + self.client.get_file("file-9", str(out), ZBGetFileOptions(download_type="phase_1", activity_data=False)) + recorded = self._only() + self.assertEqual(recorded.query["download_type"], "phase_1") + self.assertEqual(recorded.query["activity_data"], "false") + self.assertIn("client-key", out.read_text()) + + self.transport.requests.clear() + self.client.scoring_get_file("file-9", str(out) + ".s", ZBGetFileOptions(download_type="combined", activity_data=True)) + recorded = self._only() + self.assertEqual(recorded.query["download_type"], "combined") + self.assertNotIn("activity_data", recorded.query) + + def test_invalid_arguments_send_no_request(self): + csv = self._csv() + attempts = [ + lambda: ZeroBounce(" "), + lambda: ZeroBounce("k", base_url="http://evil.example/v2"), + lambda: self.client.validate("a@b.com", timeout=2), + lambda: self.client.validate("a@b.com", timeout=61), + lambda: self.client.validate_batch([]), + lambda: self.client.validate_batch([ZBValidateBatchElement("a@b.com")], timeout=9), + lambda: ZBValidateBatchElement(" "), + lambda: self.client.file_status(" "), + lambda: self.client.get_file(" ", "out.csv"), + lambda: self.client.delete_file(" "), + lambda: self.client.find_email_format("Ada"), + lambda: self.client.find_email_format("Ada", domain="example.com", company_name="Acme"), + lambda: self.client.find_domain(), + lambda: self.client.find_domain(domain="example.com", company_name="Acme"), + lambda: self.client.add_filter("", "domain", "example.com"), + lambda: self.client.delete_filter("allow", "", "example.com"), + lambda: self.client.validate_s3file(" "), + lambda: self.client.export_s3file(" ", "bucket"), + lambda: self.client.export_s3file("file-9", " "), + lambda: self.client.get_list_evaluator_result(" "), + lambda: self.client.email_finder_file_status(" "), + ] + for attempt in attempts: + self.transport.requests.clear() + with self.assertRaises((ZBClientException, ZBApiException)): + attempt() + self.assertEqual(self.transport.requests, [], attempt) + + def test_account_errors_raise_a_short_message(self): + csv = self._csv() + out = Path(tempfile.mkdtemp()) / "out.csv" + client = ZeroBounce("invalid-key") + credits = client.get_credits() + self.assertEqual(credits.credits, "-1") + + raised = [ + lambda: client.get_api_usage(date(2024, 1, 1), date(2024, 1, 2)), + lambda: client.get_activity("a@b.com"), + lambda: client.score_email("a@b.com"), + lambda: client.score_email("a@b.com", use_post=True), + lambda: client.validate("a@b.com"), + lambda: client.validate("a@b.com", use_post=True), + lambda: client.validate_batch([ZBValidateBatchElement("a@b.com")]), + lambda: client.file_status("file-9"), + lambda: client.get_file("file-9", str(out)), + lambda: client.delete_file("file-9"), + lambda: client.send_file(str(csv), 1), + lambda: client.list_filters(), + lambda: client.add_filter("block", "domain", "x.com"), + lambda: client.list_s3files(), + lambda: client.find_email_format("Ada", domain="example.com"), + lambda: client.find_domain(domain="example.com"), + lambda: client.guess_format("example.com"), + lambda: client.get_list_evaluator_result("file-9"), + lambda: client.email_finder_file_status("file-9"), + lambda: client.email_finder_get_file("file-9", str(out) + ".f"), + lambda: client.domain_search_get_file("file-9", str(out) + ".d"), + lambda: ZeroBounce("nocredits-key").validate("a@b.com"), + ] + for call in raised: + with self.assertRaises(ZBApiException) as caught: + call() + message = str(caught.exception) + self.assertGreater(len(message), 0, call) + self.assertLess(len(message), 300, call) + lowered = message.lower() + self.assertNotIn(" Date: Mon, 19 Oct 2026 06:35:19 +0400 Subject: [PATCH 7/7] fix: save plain-text file downloads and lint the SDK get_file treated every text/plain body as an error, so a CSV with that content type never saved. Empty downloads still raise. CI runs a narrow Ruff check with the unit tests. Co-authored-by: Cursor --- .github/workflows/publish.yml | 7 +++++-- .github/workflows/sdk_ci.yml | 7 +++++-- AGENTS.md | 13 +++++++++---- pyproject.toml | 3 +++ src/zerobouncesdk/zerobouncesdk.py | 4 ++-- tests/__main__.py | 1 + tests/test_offline_api.py | 1 - 7 files changed, 25 insertions(+), 11 deletions(-) diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 286c8c0..7ac54ae 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -86,8 +86,11 @@ jobs: - name: Install dependencies run: pip install -e ".[test]" - - name: Compile - run: python -m compileall -q src tests + - name: Lint + run: | + pip install ruff + ruff check src tests + python -m compileall -q src tests - name: Run tests run: | diff --git a/.github/workflows/sdk_ci.yml b/.github/workflows/sdk_ci.yml index f0b2ae7..7d4ec91 100644 --- a/.github/workflows/sdk_ci.yml +++ b/.github/workflows/sdk_ci.yml @@ -27,8 +27,11 @@ jobs: - name: Install dependencies run: pip install -e ".[test]" - - name: Compile - run: python -m compileall -q src tests + - name: Lint + run: | + pip install ruff + ruff check src tests + python -m compileall -q src tests - name: Run unit tests with coverage run: | diff --git a/AGENTS.md b/AGENTS.md index 847282d..5f0546b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -19,11 +19,15 @@ Guidance for AI coding agents working in this repository. ```bash pip install -e ".[test]" -python -m compileall -q src tests # syntax check; there is no separate typechecker or linter +pip install ruff +ruff check src tests # undefined names and syntax errors only +python -m compileall -q src tests python -m pytest # unit tests and tests/test_offline_api.py ``` -CI (`.github/workflows/sdk_ci.yml`) compiles, then runs `python -m pytest` with coverage. There is no MyPy or Ruff config. Do not add one that fails the existing tree unless asked. +`ruff check` plus `compileall` is the lint step. There is no typechecker. The Ruff selection in `pyproject.toml` is `E9`, `F63`, `F7`, and `F82`. Do not widen it to a style ruleset that fails the existing tree. + +CI (`.github/workflows/sdk_ci.yml`) runs `ruff check`, `compileall`, then `python -m pytest` with coverage. ## Conventions @@ -33,7 +37,8 @@ CI (`.github/workflows/sdk_ci.yml`) compiles, then runs `python -m pytest` with - **Booleans on form and multipart bodies must be the strings `true` and `false`.** A Python `bool` passed to `requests` becomes `True` or `False`, and the API will not treat that as a boolean. `_form_data` converts them. JSON bodies (`validate_batch`) keep JSON booleans. - **`allow_phase_2` is validation `send_file` and `validate_s3file` only.** Scoring `send_file` and `score_s3file` do not send it. `activity_data` on get-file is validation `get_file` only. - **`validate_batch` uses `api.zerobounce.net`**, not `bulkapi.zerobounce.net`. It follows `base_url`. -- **Failures raise.** `ZBClientException` means the arguments were rejected and no request was sent. `ZBApiException` means the API failed. The message must be short: no raw HTML, no megabyte body, no `JSONDecodeError` / `Expecting value` text, and no `Error: Error:` prefix. An account-level `validate_batch` error (`email_address` `all`) raises. A per-address error stays on the result's `errors` list. +- **Failures raise.** `ZBClientException` means the arguments were rejected and no request was sent. `ZBApiException` means the API failed. The message must be short: no raw HTML, no megabyte body, no `JSONDecodeError` / `Expecting value` text, and no `Error: Error:` prefix. An account-level `validate_batch` error (`email_address` `all`) raises. A per-address error stays on the result's `errors` list. `get_credits` with an invalid key returns `Credits` `-1`; that endpoint does not send an `error` field. +- **A file download is not JSON.** `get_file` saves a non-empty CSV body, including `text/csv` and `text/plain`. An empty body, an HTML page (status 200 or 403), and a JSON error are `ZBApiException`. Do not treat every `text/plain` body as an error. - **Do not follow redirects.** Every `requests` call passes `allow_redirects=False`. A 3xx response is a `ZBApiException`. A redirect must not be fetched, because the `Location` could be another host and the next request can carry the API key. - **Do not commit `.env`.** It is gitignored and may hold a real key. Tests must not call the live API. - Match the surrounding style. Do not reformat unrelated code. @@ -53,5 +58,5 @@ Publishing and pushing tags need a maintainer's explicit approval. Do not bump t 1. Move `[Unreleased]` entries under `## [x.y.z] - YYYY-MM-DD`, leave an empty `[Unreleased]` section, and update the compare links at the bottom of `CHANGELOG.md`. 2. Set the same version in `pyproject.toml` (`version = "x.y.z"`). The publish workflow fails if it differs from the tag. That is the only version field. There is no runtime `__version__` and no user-agent constant. `setup.py` does not contain a version. -3. `python -m compileall -q src tests` and `python -m pytest`. +3. `ruff check src tests`, `python -m compileall -q src tests`, and `python -m pytest`. 4. Tag `vx.y.z` on that commit and run the Publish workflow (`.github/workflows/publish.yml`) with that tag. It publishes `zerobouncesdk` to PyPI. diff --git a/pyproject.toml b/pyproject.toml index 27594a6..f5bc8a9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,9 @@ test = ["pytest>=7.0.0", "pytest-cov>=4.0.0"] [project.urls] Homepage = "https://github.com/zerobounce/zero-bounce-python-sdk-setup" +[tool.ruff.lint] +select = ["E9", "F63", "F7", "F82"] + [tool.pytest.ini_options] testpaths = ["tests"] python_files = ["test_*.py", "zero_bounce_test_case.py"] diff --git a/src/zerobouncesdk/zerobouncesdk.py b/src/zerobouncesdk/zerobouncesdk.py index c0283d0..caad7b4 100644 --- a/src/zerobouncesdk/zerobouncesdk.py +++ b/src/zerobouncesdk/zerobouncesdk.py @@ -284,12 +284,12 @@ def _raise_if_download_error(self, response, body_str: str, content_type: str) - raise ZBApiException( self._redact("ZeroBounce returned a redirect, which this client does not follow.") ) + if not (body_str or "").strip(): + raise ZBApiException(self._redact(_message_from_body(body_str, status))) if ( status > 299 or _looks_like_html(body_str, content_type) or self._should_treat_get_file_body_as_error(body_str, content_type) - or not body_str.strip() - or "text/plain" in content_type.lower() ): raise ZBApiException(self._redact(_message_from_body(body_str, status))) diff --git a/tests/__main__.py b/tests/__main__.py index 8196143..72e7a10 100644 --- a/tests/__main__.py +++ b/tests/__main__.py @@ -1,5 +1,6 @@ import unittest +from .test_offline_api import OfflineApiTest from .zero_bounce_test_case import ZeroBounceTestCase from .zero_bounce_integration_test import ZeroBounceIntegrationTestCase diff --git a/tests/test_offline_api.py b/tests/test_offline_api.py index bb8283b..9154bd6 100644 --- a/tests/test_offline_api.py +++ b/tests/test_offline_api.py @@ -19,7 +19,6 @@ from unittest.mock import patch from urllib.parse import parse_qs, urlparse -import requests from requests import Response from zerobouncesdk import (