Bulk-Verify a CSV of Emails with Python

This script takes a CSV with an email column, submits it as a single batch job, polls for completion, and writes output.csv with the original rows plus status, score, and reason columns appended.

Requirements

pip install requests

The script

import csv
import os
import sys
import time
import requests

API_KEY = os.environ["ROCKETVERIFIER_API_KEY"]  # rv_live_... or rv_test_...
BASE_URL = "https://api.rocketverifier.com"
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}


def read_emails(path: str) -> list[str]:
    with open(path, newline="") as f:
        reader = csv.DictReader(f)
        return [row["email"].strip() for row in reader if row.get("email", "").strip()]


def create_batch_job(emails: list[str]) -> str:
    resp = requests.post(
        f"{BASE_URL}/v1/verify/batch",
        headers=HEADERS,
        json={"emails": emails},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["data"]["jobId"]


def poll_job(job_id: str, interval_seconds: int = 5) -> None:
    while True:
        resp = requests.get(f"{BASE_URL}/v1/verify/job/{job_id}", headers=HEADERS, timeout=30)
        resp.raise_for_status()
        status = resp.json()["data"]["status"]
        print(f"Job {job_id}: {status}")
        if status == "completed":
            return
        if status == "failed":
            raise RuntimeError(f"Batch job {job_id} failed")
        time.sleep(interval_seconds)


def fetch_results(job_id: str) -> list[dict]:
    resp = requests.get(f"{BASE_URL}/v1/verify/job/{job_id}/results", headers=HEADERS, timeout=30)
    resp.raise_for_status()
    return resp.json()["data"]["results"]


def write_output(results: list[dict], path: str = "output.csv") -> None:
    with open(path, "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["email", "status", "score", "reason", "is_catch_all", "is_disposable"])
        for r in results:
            writer.writerow([
                r["email"], r["status"], r["score"], r["reason"],
                r["domain"]["acceptAll"], r["domain"]["disposable"],
            ])
    print(f"Wrote {len(results)} rows to {path}")


def main():
    input_path = sys.argv[1] if len(sys.argv) > 1 else "emails.csv"
    emails = read_emails(input_path)
    print(f"Submitting {len(emails)} emails...")
    job_id = create_batch_job(emails)
    poll_job(job_id)
    results = fetch_results(job_id)
    write_output(results)


if __name__ == "__main__":
    main()

Running it against the sandbox first

export ROCKETVERIFIER_API_KEY=rv_test_your_sandbox_key
python3 verify_csv.py emails.csv

Point emails.csv at a file of @sandbox.rocketverifier.com addresses to see the full flow (submit → poll → results → CSV output) without spending real credits, then swap in a live key and real data once you’ve confirmed the output looks right.

Error handling notes

requests.raise_for_status() will throw on any non-2xx response, including a 401 for an invalid API key or 402 for insufficient credits. Wrap the create_batch_job call in a try/except requests.HTTPError block in production and inspect resp.json() for the error detail before retrying.