Email verification in Python

import re
import smtplib
import dns.resolver                                   # pip install dnspython
from email_validator import validate_email, EmailNotValidError   # pip install email-validator

# 1. Shape only. Fine for catching typos in a form field.
SHAPE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+
quot;)

# 2. Real syntax rules plus a DNS check that the domain accepts mail.
def check_syntax(email: str) -> str | None:
    try:
        info = validate_email(email, check_deliverability=True)
        return info.normalized
    except EmailNotValidError as exc:
        print(f"rejected: {exc}")
        return None

# 3. The SMTP handshake by hand. Quits before DATA, so nothing is sent.
def smtp_probe(email: str) -> int:
    domain = email.rsplit("@", 1)[1]
    mx = min(dns.resolver.resolve(domain, "MX"), key=lambda r: r.preference)
    with smtplib.SMTP(str(mx.exchange).rstrip("."), 25, timeout=10) as smtp:
        smtp.ehlo("verify.example.com")
        smtp.mail("probe@verify.example.com")
        code, _ = smtp.rcpt(email)   # 250 accepted, 550 no such user, 4xx try later
    return code

pip install requests

import os
import requests

session = requests.Session()
session.headers.update({"X-API-Key": os.environ["ZAPBOUNCE_KEY"]})

def verify(email: str) -> dict:
    r = session.post(
        "https://app.zapbounce.com/api/v1/verify",
        json={"emails": [email]},
        timeout=(3.05, 10),   # connect, read — the read side is the slow mail server
    )
    r.raise_for_status()
    return r.json()["results"][0]

result = verify("ada@example.com")

if result["status"] == "valid":
    send_to(result["email"])
elif result["status"] == "catch_all":
    queue_for_separate_send(result["email"])   # the domain accepts everything
elif result["status"] == "unknown":
    keep_and_retry_later(result["email"])      # credit refunded, not a verdict
else:
    suppress(result["email"])

if "disposable" in result["flags"]:
    suppress(result["email"])

import time
from typing import Iterator

API = "https://app.zapbounce.com/api/v1"
DONE = {"completed", "partial", "failed", "cancelled"}

def submit(emails: list[str]) -> str:
    r = session.post(
        f"{API}/jobs",
        json={"emails": emails, "source": "api"},   # up to 50,000 per job
        timeout=(3.05, 30),
    )
    r.raise_for_status()
    return r.json()["id"]

def wait(job_id: str, every: float = 5.0) -> dict:
    """Poll the job until it reaches a terminal status. There is no webhook."""
    while True:
        job = session.get(f"{API}/jobs/{job_id}", timeout=(3.05, 30)).json()
        if job["status"] in DONE:
            return job
        time.sleep(every)

def rows(job_id: str) -> Iterator[dict]:
    """Yield every row, page by page. Never build the full list in memory."""
    page = 1
    while True:
        body = session.get(
            f"{API}/jobs/{job_id}/emails",
            params={"page": page, "pageSize": 100}, timeout=(3.05, 30),
        ).json()
        yield from body["items"]
        if page >= body["meta"]["totalPages"]:
            return
        page += 1

job_id = submit(load_emails("contacts.csv"))
job = wait(job_id)
print(f"{job['validEmails']} valid, {job['invalidEmails']} invalid, "
      f"{job['catchAllEmails']} catch-all, {job['otherEmails']} unknown (refunded)")
for row in rows(job_id):
    handle(row)

import random
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# The API sends no Retry-After header, so urllib3 only retries 5xx here.
# 429 is handled below, using X-RateLimit-Window-Seconds.
retry = Retry(
    total=3,
    backoff_factor=0.5,
    status_forcelist=[500, 502, 503, 504],
    allowed_methods=["GET", "POST"],
)
session.mount("https://", HTTPAdapter(max_retries=retry, pool_maxsize=20))

def verify_with_backoff(email: str, attempts: int = 4) -> dict:
    for attempt in range(attempts):
        try:
            return verify(email)
        except requests.HTTPError as exc:
            resp = exc.response
            body = resp.json()   # {"statusCode", "code", "message", "requestId"}
            if body.get("code") == "CREDITS_INSUFFICIENT":
                raise OutOfCredits(body["message"]) from exc      # 422, retrying will not help
            if resp.status_code == 429 and attempt < attempts - 1:
                window = float(resp.headers.get("X-RateLimit-Window-Seconds", 60))
                time.sleep(window * (2 ** attempt) / 4 + random.random())
                continue
            logger.warning("verify failed request_id=%s", body.get("requestId"))
            raise
        except requests.Timeout:
            # A verifier timing out is not a reason to block a signup. Fail open.
            return {"email": email, "status": "unknown", "reasonCode": "TIMEOUT", "flags": []}
    raise RuntimeError("rate limited")