Email verification in Django
Django's validator hook is the natural place for a signup check, with one rule: it must fail open. A validator that raises ValidationError when the API times out rejects a real customer because of a slow socket, and you will never see the lost registration in your logs.
For anything list-shaped, a Celery task submits a job and a second task polls it until it finishes. No management command running under nohup.
requests with a module-level Session, wired into a django.core.validators callable and a Celery task.
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.
# accounts/validators.py
import logging
import requests
from django.conf import settings
from django.core.exceptions import ValidationError
logger = logging.getLogger(__name__)
_session = requests.Session()
_session.headers["X-API-Key"] = settings.ZAPBOUNCE_KEY
# catch_all and unknown are not rejections. Only a definite no blocks a signup,
# plus the disposable flag, which is an entry in the flags array beside the status.
def deliverable_email(value: str) -> None:
"""Field validator. Fails OPEN: a slow verifier must not cost a signup."""
try:
response = _session.post(
"https://app.zapbounce.com/api/v1/verify",
json={"emails": [value]},
timeout=(2, 3),
)
response.raise_for_status()
except requests.RequestException:
logger.warning("zapbounce unreachable, allowing %s", value)
return
result = response.json()["results"][0]
if result["status"] == "invalid" or "disposable" in result["flags"]:
raise ValidationError("That address will not receive mail. Check it for a typo.")
# accounts/forms.py
class SignupForm(forms.Form):
email = forms.EmailField(validators=[deliverable_email])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.
# contacts/tasks.py
from celery import shared_task
from django.db import transaction
from .models import Contact, ContactList
API = "https://app.zapbounce.com/api/v1"
DONE = {"completed", "partial", "failed", "cancelled"}
@shared_task(bind=True, max_retries=3, retry_backoff=True, retry_jitter=True)
def verify_list(self, list_id: int) -> str:
emails = list(
Contact.objects.filter(list_id=list_id, email_result="")
.values_list("email", flat=True)
.iterator(chunk_size=5000) # do not materialise a million rows
)
response = _session.post(
f"{API}/jobs",
json={"emails": emails, "source": "api"}, # up to 50,000 per job
headers={"Idempotency-Key": f"list-{list_id}-{self.request.id}"},
timeout=(3, 60),
)
response.raise_for_status()
job_id = response.json()["id"]
with transaction.atomic():
ContactList.objects.filter(pk=list_id).update(job_id=job_id)
poll_job.apply_async(args=[job_id], countdown=30)
return job_id
@shared_task(bind=True, max_retries=None)
def poll_job(self, job_id: str) -> None:
"""No webhook exists. Re-queue this task until the job reaches a terminal status."""
job = _session.get(f"{API}/jobs/{job_id}", timeout=(3, 30)).json()
if job["status"] not in DONE:
raise self.retry(countdown=30)
import_results.delay(job_id)
@shared_task
def import_results(job_id: str) -> None:
"""Bulk-updates rather than saving row by row."""
updates, page = [], 1
while True:
body = _session.get(
f"{API}/jobs/{job_id}/emails",
params={"page": page, "pageSize": 100}, timeout=(3, 30),
).json()
updates += [Contact(email=r["email"], email_result=r["status"]) for r in body["items"]]
if page >= body["meta"]["totalPages"]:
break
page += 1
Contact.objects.bulk_update(updates, ["email_result"], batch_size=1000)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.
# contacts/zapbounce.py
import random
import time
import requests
class OutOfCredits(Exception):
pass
def verify_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=(2, 10),
)
except requests.Timeout:
# A timeout is not a bad address. Fail open.
return {"email": email, "status": "unknown", "reasonCode": "TIMEOUT", "flags": []}
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 will not help
if response.status_code == 429:
# There is 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
logger.warning("zapbounce %s request_id=%s", err["code"], err["requestId"])
response.raise_for_status()
raise RuntimeError("rate limited")Django: common questions
Should the validator block on unknown?
Never. Unknown means a mail server refused to answer, which says nothing about the person filling in your form.
Do I need Celery?
Something off the request cycle, yes. Celery, RQ or Django-Q all work; the jobs endpoint plus a task that retries itself means each task is short whichever you pick.
How do I test the unknown branch?
Mock the response in your test with a result whose status is unknown. A healthy address never produces one, so forcing it is the only way to hit the branch that never fires locally.
Run this Django code today
100 free checks a month, no card, credits that never expire, and unknown results refunded.