Email verification in Flask

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

session = requests.Session()


def init_app(app):
    session.headers["Authorization"] = f"Bearer {app.config['ZAPBOUNCE_KEY']}"


def check(email: str) -> dict:
    """Return the response body, 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://api.zapbounce.com/v1/verify",
            json={"email": email},
            timeout=(1.5, 2.0),
        )
        response.raise_for_status()
        return response.json()
    except requests.RequestException:
        current_app.logger.warning("zapbounce unreachable for %s", email)
        return {"result": "unknown", "disposable": False}


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

    # disposable is a flag beside the result, so it gets its own test.
    if checked["result"] == "invalid" or checked["disposable"]:
        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["result"])
    return redirect(url_for("main.welcome", user_id=user.id))

# app/jobs.py
from rq import Retry
from rq.decorators import job

from . import queue
from .verification import session


@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(
        "https://api.zapbounce.com/v1/batches",
        json={
            "name": f"list-{list_id}",
            "emails": emails,
            "webhook_url": current_app.config["ZAPBOUNCE_WEBHOOK_URL"],
        },
        headers={"Idempotency-Key": f"list-{list_id}"},
        timeout=(3, 60),
    )
    response.raise_for_status()
    return response.json()["batch_id"]


@job(queue)
def import_results(batch_id: str) -> None:
    cursor, seen = None, 0
    while True:
        params = {"limit": 1000, **({"cursor": cursor} if cursor else {})}
        page = session.get(
            f"https://api.zapbounce.com/v1/batches/{batch_id}/results",
            params=params, timeout=(3, 30),
        ).json()

        db.session.bulk_update_mappings(Contact, [
            {"email": r["email"], "email_result": r["result"]} for r in page["data"]
        ])
        db.session.commit()          # commit per page, so a crash loses one page
        seen += len(page["data"])

        if not page["has_more"]:
            break
        cursor = page["next_cursor"]

# app/webhooks.py
import hashlib
import hmac
import time

from flask import Blueprint, abort, request

bp = Blueprint("webhooks", __name__)


@bp.post("/webhooks/zapbounce")
def zapbounce():
    raw = request.get_data()            # cache=True by default: safe to call again
    header = request.headers.get("ZapBounce-Signature", "")
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)

    if abs(time.time() - int(parts.get("t", 0))) > 300:
        abort(401, "stale signature")

    expected = hmac.new(
        current_app.config["ZAPBOUNCE_WEBHOOK_SECRET"].encode(),
        f"{parts['t']}.".encode() + raw,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected, parts.get("v1", "")):
        abort(401, "bad signature")

    event = request.get_json()

    if not WebhookEvent.query.get(event["id"]):
        db.session.add(WebhookEvent(id=event["id"]))
        db.session.commit()
        import_results.delay(event["data"]["batch_id"])

    return "", 204                      # acknowledge now, import on the worker