How to verify a CSV email list through an API

To verify an email list with an API, run a job that reads the CSV, sends each address from your server and writes the result beside its source row. Save progress as you go: a request can finish immediately, remain pending or fail before producing an address verdict.

This tutorial includes a Python script for that job. You keep the file and manage the run; Emvero receives one address per verification request. The bulk verification page explains the service’s list workflow. There is no hosted CSV uploader or batch endpoint.

Prepare the file and the API credential

Use a CSV with a header and an email column. The sample below is synthetic; the quoted comma belongs to the note, and the last row has no address:

contact_id,email,note
c-001,ada@example.com,"trial, invited"
c-002,ops@example.org,newsletter
c-003,,missing address

Keep the original file and choose a separate output path. The script preserves its columns and the original address value. It strips surrounding whitespace from the copy it submits, without lowercasing or otherwise rewriting the address. An empty address becomes an input error attached to that row.

You need Python 3, an API credential in EMVERO_API_TOKEN and permission to create the output and checkpoint files. Choose a conservative worker limit for your account. The sample loads the CSV into memory; for a file that does not fit, adapt the job to read and persist smaller groups.

Requests go to https://api.emvero.tech/v1/verifications with Authorization: Bearer. Run the code locally or on a server so the credential stays out of browser code. See the email verification API for the request contract and security page for data handling.

Keep a stable identity for the job and each row

The script hashes the input file’s bytes together with the selected email-column name. That job identifier and the row’s position form emvero_row_id and a stable Idempotency-Key. Repeating the unchanged input produces the same keys. Editing the file, reordering it or choosing another column produces a different job, so old checkpoint entries cannot be assigned to new rows.

Within the supported idempotency window, repeating a key retrieves the original operation instead of starting another verification. This recipe does not depend on a particular window duration. It also saves the ID of each pending verification. On restart, it retrieves that ID before considering another POST.

Each source row is a separate job item. Duplicate addresses in different rows receive different keys; the script does not deduplicate the list. If you already have immutable record and job IDs, you can use those to define your own identity scheme.

Write evidence and errors into separate columns

The output keeps the input columns and adds the fields below. Structured objects remain compact JSON inside a CSV cell, so nested fields and reason-code arrays survive the round trip.

If the next system accepts JSON or database records, keep these objects in their native form. Flattening them into today’s known check columns can silently discard fields added later.

Run the Python job

Save the following script as verify_csv.py. Python’s csv parser handles quoted fields and empty cells. The script rejects duplicate headers and reserved result-column names before processing, rather than overwriting input values.

The checkpoint is an append-only JSON Lines file. On resume, the latest event for each row wins. Terminal results are reused and pending IDs are retrieved; a resumed job does not reverify every completed row with fresh evidence.

#!/usr/bin/env python3
import argparse
import csv
import hashlib
import json
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen

API_BASE = "https://api.emvero.tech"
TERMINAL_STATUSES = {"deliverable", "risky", "undeliverable", "unknown"}
RETRYABLE_HTTP_STATUSES = {429, 500, 503}
MAX_REQUEST_ATTEMPTS = 4

RESULT_COLUMNS = [
    "emvero_row_id",
    "emvero_submitted_email",
    "emvero_state",
    "emvero_verification_id",
    "emvero_status",
    "emvero_action",
    "emvero_checks",
    "emvero_reason_codes",
    "emvero_confidence",
    "emvero_freshness",
    "emvero_normalized",
    "emvero_error_http_status",
    "emvero_error_code",
    "emvero_error_message",
]

checkpoint_lock = threading.Lock()


class ApiError(Exception):
    def __init__(self, http_status, code, message, retry_after_seconds=0):
        super().__init__(message or code)
        self.http_status = http_status
        self.code = code
        self.message = message
        self.retry_after_seconds = retry_after_seconds


def retry_after_seconds(value):
    if not value:
        return 0
    value = value.strip()
    if value.isdigit():
        return int(value)
    try:
        return max(0, (parsedate_to_datetime(value) - datetime.now(timezone.utc)).total_seconds())
    except (TypeError, ValueError):
        return 0


def compact_json(value):
    if value is None:
        return ""
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))


def request_json(method, path, token, body=None, idempotency_key=None):
    headers = {
        "Accept": "application/json",
        "Authorization": f"Bearer {token}",
    }
    data = None

    if body is not None:
        data = json.dumps(body).encode("utf-8")
        headers["Content-Type"] = "application/json"
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key

    request = Request(
        f"{API_BASE}{path}",
        data=data,
        headers=headers,
        method=method,
    )

    try:
        with urlopen(request, timeout=30) as response:
            return json.load(response)
    except HTTPError as error:
        raw = error.read()
        try:
            payload = json.loads(raw)
        except (json.JSONDecodeError, UnicodeDecodeError):
            payload = {}
        raise ApiError(
            error.code,
            payload.get("error", "http_error"),
            payload.get("message", ""),
            retry_after_seconds(error.headers.get("Retry-After")),
        ) from error


def call_with_backoff(method, path, token, body=None, idempotency_key=None):
    for attempt in range(MAX_REQUEST_ATTEMPTS):
        delay = min(2**attempt, 8)
        try:
            return request_json(method, path, token, body, idempotency_key)
        except ApiError as error:
            retryable = error.http_status in RETRYABLE_HTTP_STATUSES
            if not retryable or attempt == MAX_REQUEST_ATTEMPTS - 1:
                raise
            delay = max(delay, error.retry_after_seconds)
        except (URLError, TimeoutError):
            if attempt == MAX_REQUEST_ATTEMPTS - 1:
                raise

        # Keep local retries bounded, and honour a longer server Retry-After.
        time.sleep(delay)


def load_checkpoint(path):
    latest = {}
    if not path.exists():
        return latest

    with path.open("rb") as handle:
        for line in handle:
            try:
                event = json.loads(line)
                latest[event["row_id"]] = event
            except (json.JSONDecodeError, UnicodeDecodeError, KeyError):
                # Ignore an incomplete final line left by an interrupted write.
                continue
    return latest


def append_checkpoint(path, event):
    line = json.dumps(event, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    with checkpoint_lock:
        with path.open("a+b") as handle:
            handle.seek(0, os.SEEK_END)
            if handle.tell():
                handle.seek(-1, os.SEEK_END)
                if handle.read(1) != b"\n":
                    handle.write(b"\n")
            handle.write(line + b"\n")
            handle.flush()
            os.fsync(handle.fileno())


def row_id(index, job_id):
    return f"{job_id}-row-{index:08d}"


def idempotency_key(job_id, item_id):
    return f"csv-{item_id}"


def read_input(path, email_column, job_id):
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        if not reader.fieldnames:
            raise ValueError("the CSV file has no header row")
        if len(reader.fieldnames) != len(set(reader.fieldnames)):
            raise ValueError("the CSV file has duplicate column names")
        if email_column not in reader.fieldnames:
            raise ValueError(f"the CSV file has no {email_column!r} column")
        reserved = set(reader.fieldnames) & set(RESULT_COLUMNS)
        if reserved:
            raise ValueError(f"input already has result columns: {sorted(reserved)}")

        items = []
        for index, source_row in enumerate(reader, start=1):
            if None in source_row:
                raise ValueError(
                    f"data row {index} has more fields than the header"
                )
            raw_email = source_row.get(email_column) or ""
            items.append(
                {
                    "row_id": row_id(index, job_id),
                    "source_row": source_row,
                    "submitted_email": raw_email.strip(),
                }
            )
        return list(reader.fieldnames), items


def pending_event(item_id, key, result):
    return {
        "row_id": item_id,
        "state": "pending",
        "idempotency_key": key,
        "verification_id": result.get("id", ""),
        "retry_after_ms": result.get("retry_after_ms", 1000),
    }


def terminal_event(item_id, key, result):
    return {
        "row_id": item_id,
        "state": "terminal",
        "idempotency_key": key,
        "result": result,
    }


def poll_until_terminal(result, token, poll_seconds, checkpoint_path, item_id, key):
    deadline = time.monotonic() + poll_seconds

    while result.get("status") == "pending":
        verification_id = result.get("id")
        if not verification_id:
            raise ValueError("a pending result has no verification id")

        append_checkpoint(checkpoint_path, pending_event(item_id, key, result))
        delay_ms = result.get("retry_after_ms") or 1000
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return result, False

        time.sleep(min(max(delay_ms / 1000, 0.25), remaining))
        result = call_with_backoff(
            "GET",
            f"/v1/verifications/{quote(verification_id, safe='')}",
            token,
        )

    return result, True


def verify_item(item, job_id, token, poll_seconds, checkpoint_path, saved):
    item_id = item["row_id"]
    email = item["submitted_email"]
    key = idempotency_key(job_id, item_id)

    if not email:
        return {"kind": "input_error", "code": "empty_email", "message": "email is empty"}

    previous = saved.get(item_id)
    if previous and previous.get("state") == "terminal":
        return {"kind": "terminal", "result": previous["result"]}

    result = None
    if previous and previous.get("state") == "pending" and previous.get("verification_id"):
        try:
            result = call_with_backoff(
                "GET",
                f"/v1/verifications/{quote(previous['verification_id'], safe='')}",
                token,
            )
        except ApiError as error:
            if error.http_status != 404:
                raise
            # The stored result is no longer retrievable. Reuse the same stable
            # idempotency key and let the API decide whether it still owns it.

    if result is None:
        result = call_with_backoff(
            "POST",
            "/v1/verifications",
            token,
            body={"email": email},
            idempotency_key=key,
        )

    result, finished = poll_until_terminal(
        result,
        token,
        poll_seconds,
        checkpoint_path,
        item_id,
        key,
    )

    if not finished:
        return {
            "kind": "pending",
            "result": result,
            "code": "client_poll_deadline",
            "message": "verification is still pending; resume the job later",
        }

    status = result.get("status")
    if status not in TERMINAL_STATUSES:
        return {
            "kind": "protocol_error",
            "result": result,
            "code": "unexpected_status",
            "message": f"unexpected verification status: {status!r}",
        }

    append_checkpoint(checkpoint_path, terminal_event(item_id, key, result))
    return {"kind": "terminal", "result": result}


def run_item(item, job_id, token, poll_seconds, checkpoint_path, saved):
    try:
        return verify_item(
            item,
            job_id,
            token,
            poll_seconds,
            checkpoint_path,
            saved,
        )
    except ApiError as error:
        return {
            "kind": "api_error",
            "http_status": error.http_status,
            "code": error.code,
            "message": error.message,
        }
    except (URLError, TimeoutError) as error:
        return {
            "kind": "network_error",
            "code": "network_error",
            "message": str(getattr(error, "reason", error)),
        }
    except (KeyError, TypeError, ValueError) as error:
        return {
            "kind": "protocol_error",
            "code": "invalid_response",
            "message": str(error),
        }


def output_fields(item, outcome):
    result = outcome.get("result") or {}
    return {
        "emvero_row_id": item["row_id"],
        "emvero_submitted_email": item["submitted_email"],
        "emvero_state": outcome["kind"],
        "emvero_verification_id": result.get("id", ""),
        "emvero_status": result.get("status", ""),
        "emvero_action": result.get("action", ""),
        "emvero_checks": compact_json(result.get("checks")),
        "emvero_reason_codes": compact_json(result.get("reason_codes")),
        "emvero_confidence": result.get("confidence", ""),
        "emvero_freshness": compact_json(result.get("freshness")),
        "emvero_normalized": compact_json(result.get("normalized")),
        "emvero_error_http_status": outcome.get("http_status", ""),
        "emvero_error_code": outcome.get("code", ""),
        "emvero_error_message": outcome.get("message", ""),
    }


def write_output(path, input_columns, items, outcomes):
    columns = input_columns + [name for name in RESULT_COLUMNS if name not in input_columns]
    temporary_path = path.with_name(path.name + ".tmp")

    with temporary_path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns, extrasaction="ignore")
        writer.writeheader()
        for item in items:
            row = dict(item["source_row"])
            row.update(output_fields(item, outcomes[item["row_id"]]))
            writer.writerow(row)

    os.replace(temporary_path, path)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("input_csv", type=Path)
    parser.add_argument("output_csv", type=Path)
    parser.add_argument("--email-column", default="email")
    parser.add_argument("--workers", type=int, default=4)
    parser.add_argument("--poll-seconds", type=int, default=300)
    parser.add_argument("--checkpoint", type=Path)
    args = parser.parse_args()

    if args.workers < 1:
        parser.error("--workers must be at least 1")
    if args.poll_seconds < 1:
        parser.error("--poll-seconds must be at least 1")
    if args.input_csv.resolve() == args.output_csv.resolve():
        parser.error("input_csv and output_csv must be different files")

    token = os.environ.get("EMVERO_API_TOKEN")
    if not token:
        parser.error("set EMVERO_API_TOKEN before running the job")

    checkpoint_path = args.checkpoint or Path(str(args.output_csv) + ".checkpoint.jsonl")
    if checkpoint_path.resolve() in {args.input_csv.resolve(), args.output_csv.resolve()}:
        parser.error("checkpoint must be separate from input and output")
    job_id = hashlib.sha256(
        args.input_csv.read_bytes() + b"\0" + args.email_column.encode("utf-8")
    ).hexdigest()[:20]
    input_columns, items = read_input(args.input_csv, args.email_column, job_id)
    saved = load_checkpoint(checkpoint_path)

    outcomes = {}
    with ThreadPoolExecutor(max_workers=args.workers) as executor:
        futures = {
            executor.submit(
                run_item,
                item,
                job_id,
                token,
                args.poll_seconds,
                checkpoint_path,
                saved,
            ): item["row_id"]
            for item in items
        }
        for future in as_completed(futures):
            outcomes[futures[future]] = future.result()

    write_output(args.output_csv, input_columns, items, outcomes)

    counts = {}
    for outcome in outcomes.values():
        counts[outcome["kind"]] = counts.get(outcome["kind"], 0) + 1
    print(json.dumps(counts, sort_keys=True))


if __name__ == "__main__":
    main()

Use different input and output paths:

export EMVERO_API_TOKEN="emv_your_credential"
python3 verify_csv.py contacts.csv contacts.verified.csv --workers 4

The --workers limit controls concurrent row processing. --poll-seconds controls the polling budget for a row, and --checkpoint can select the checkpoint path. Keep that path separate from both CSV files.

The script makes a bounded number of retries for 429, 500, 503 and network failures. It honours a server Retry-After longer than its local backoff. Persistent errors are written beside the row. Before resuming, fix the credential, balance or service problem, or reduce concurrency if requests are being limited.

Continue the same verification when it is pending

POST /v1/verifications returns HTTP 200 for both a terminal result and pending. For pending work, save the returned id, wait for retry_after_ms and call GET /v1/verifications/{id}. An unfinished answer is no reason to create another verification.

If the script’s polling budget expires, it records client_poll_deadline and keeps emvero_status at pending. That code describes the client’s waiting limit, not the address. Run the same command later with the same input and checkpoint to continue.

There is no outgoing callback in the current contract. Your job stores the ID and retrieves the result. A list job can accommodate that wait differently from a signup request; the real-time versus bulk comparison explains the choice.

Read an HTTP error as a job problem

A successful HTTP-200 response can contain undeliverable: the verification established an address result. A 401, 402, 403, 429, 500 or 503 is an operational response and must not be written as undeliverable or invented as unknown.

Read request errors in emvero_error_http_status, emvero_error_code and emvero_error_message. Address evidence belongs in emvero_status, emvero_action and the result fields. Once the operational cause is resolved, rerun the same job. The checkpoint retains completed results while failed rows are tried again.

Apply your list policy to the saved results

Use action as advice for the next step and read the evidence that explains it. Emvero’s recommendations support account-entry decisions; they are not permission to send a campaign. Keep subscription records, opt-outs and suppression rules in your sending system.

• With deliverable, retain the result and freshness. Observed acceptance does not guarantee future delivery or inbox placement.
• With risky, read action and reason_codes before choosing a rule. Disposable and catch-all evidence can warrant different treatment. A role-account annotation alone does not make the result risky.
• With undeliverable, exclude the address from sending when the returned evidence supports that decision. Preserve the source row and the other customer fields.
• With unknown, keep the inconclusive verdict separate from both positive and negative groups. It can still contain useful domain evidence. Decide whether to hold or recheck it; a confirmation email, if appropriate, belongs to your own workflow. Verification sends no message.

The list-cleaning guide covers sending rules. For a CRM export, the CRM email validation guide explains how to retain the customer record when its email field is unusable. The verification result guide explains the fields, including why confidence is not a probability.

Check the output before handing it on

Every input row needs a recorded state, including empty addresses and failed requests. Compare row counts, then inspect input_error, api_error, network_error, protocol_error and pending before using the output in another system.

For terminal rows, retain status, action, checks, reason_codes, confidence and freshness wherever the API supplies them. Archive the source file, job identifier, output and checkpoint together. Keep the checkpoint while you may need to resume; remove it only after the output is safely stored.

Start with the small synthetic CSV before running your own list. The bulk verification workflow can help you decide how this job fits your process; it cannot promise zero bounces or successful inbox delivery.