ZapBounce

Email verification in Flask

Flask leaves the structure to you, so the useful thing here is where each piece goes rather than how to call an API. Verification at signup belongs in a small helper with a hard timeout, and list work belongs on a worker.

One Flask-specific detail carries the worker: current_app only exists inside an app context, so a job running on an RQ worker has to push one before it reads config or logs.

requests.Session held on the app extension object, so it survives across requests and pools connections.

Verify one address

You branch on four statuses: valid, invalid, catch_all and unknown. Role and disposable arrive in the flags array beside the status, so test them separately. Unknown is the branch most code forgets, and its credit is refunded.

Python
# app/verification.py
import requests
from flask import current_app

session = requests.Session()

def init_app(app):
    session.headers["X-API-Key"] = app.config["ZAPBOUNCE_KEY"]

def check(email: str) -> dict:
    """Return the result, or an 'unknown' stand-in if we ran out of time.
    Two seconds is the budget. Someone is staring at a form.
    """
    try:
        response = session.post(
            "https://app.zapbounce.com/api/v1/verify",
            json={"emails": [email]},
            timeout=(1.5, 2.0),
        )
        response.raise_for_status()
        return response.json()["results"][0]
    except requests.RequestException:
        current_app.logger.warning("zapbounce unreachable for %s", email)
        return {"email": email, "status": "unknown", "reasonCode": "TIMEOUT", "flags": []}

# app/routes.py
@bp.post("/signup")
def signup():
    email = request.form["email"]
    checked = check(email)

    # disposable is a flag beside the status, so it gets its own test.
    if checked["status"] == "invalid" or "disposable" in checked["flags"]:
        flash("That address will not receive mail. Check it for a typo.")
        return redirect(url_for("main.signup"))

    # valid, catch_all and unknown all proceed. Only a definite no blocks.
    user = User.create(email=email, email_result=checked["status"])
    return redirect(url_for("main.welcome", user_id=user.id))

Verify a list

Submit a job rather than looping the single endpoint. We pace probes per receiving mail host, which protects the sending reputation your results depend on.

Python
# app/jobs.py
import time
from rq import Retry
from rq.decorators import job
from . import queue
from .verification import session

API = "https://app.zapbounce.com/api/v1"

@job(queue, retry=Retry(max=3, interval=[30, 120, 600]))
def verify_list(list_id: int) -> str:
    emails = [row.email for row in Contact.query.filter_by(list_id=list_id, email_result=None)]
    response = session.post(
        f"{API}/jobs",
        json={"emails": emails, "source": "api"},      # up to 50,000 per job
        headers={"Idempotency-Key": f"list-{list_id}"},
        timeout=(3, 60),
    )
    response.raise_for_status()
    job_id = response.json()["id"]
    import_results.delay(job_id)
    return job_id

@job(queue, timeout=3600)
def import_results(job_id: str) -> None:
    # No webhook exists: poll the job, then read the rows page by page.
    while session.get(f"{API}/jobs/{job_id}", timeout=(3, 30)).json()["status"] not in (
        "completed", "partial", "failed", "cancelled"
    ):
        time.sleep(10)

    page = 1
    while True:
        body = session.get(
            f"{API}/jobs/{job_id}/emails",
            params={"page": page, "pageSize": 100}, timeout=(3, 30),
        ).json()
        db.session.bulk_update_mappings(Contact, [
            {"email": r["email"], "email_result": r["status"]} for r in body["items"]
        ])
        db.session.commit()          # commit per page, so a crash loses one page
        if page >= body["meta"]["totalPages"]:
            break
        page += 1

Errors and retries

A 429 and a 5xx are worth retrying. A 422 CREDITS_INSUFFICIENT and a 400 for a malformed address will fail the same way on every attempt, so stop.

Python
# app/verification.py
import random
import time

class OutOfCredits(Exception):
    pass

def check_with_backoff(email: str, attempts: int = 4) -> dict:
    for attempt in range(attempts):
        try:
            response = session.post(
                "https://app.zapbounce.com/api/v1/verify",
                json={"emails": [email]},
                timeout=(1.5, 5),
            )
        except requests.Timeout:
            return {"email": email, "status": "unknown", "reasonCode": "TIMEOUT", "flags": []}  # fail open

        if response.ok:
            return response.json()["results"][0]

        err = response.json()      # {"statusCode", "code", "message", "requestId"}
        if err["code"] == "CREDITS_INSUFFICIENT":
            raise OutOfCredits(err["message"])    # 422, retrying changes nothing
        if response.status_code == 429:
            # No Retry-After header: back off from the window length.
            window = float(response.headers.get("X-RateLimit-Window-Seconds", 60))
            time.sleep(window * (2 ** attempt) / 4 + random.random())
            continue
        current_app.logger.warning("zapbounce %s requestId=%s", err["code"], err["requestId"])
        response.raise_for_status()
    raise RuntimeError("rate limited")

Flask: common questions

Should signup block on catch-all?

No. The domain accepts every address by design, which tells you nothing about this person. Record the verdict and let them in.

RQ or Celery?

Either. RQ is here because the retry decorator is two lines and Flask projects often already have Redis.

How do I test the unknown branch?

Monkeypatch check to return a result whose status is unknown. A healthy address never produces one, so force it in a test and confirm the signup still goes through.

API reference

Run this Flask code today

100 free checks a month, no card, credits that never expire, and unknown results refunded.

Get an API keySee pricing