---
title: "Verify a CSV Email List With an API: Python Workflow | Emvero"
description: "Build a CSV-to-API email verification job in Python. Keep source rows, resume pending checks and save results and errors without a hosted uploader."
url: https://emvero.io/blog/validate-email-list-csv-api/
language: en
updated: 2026-10-10
---

Blog

# How to verify a CSV email list through an API

Read a CSV safely, send one server-side verification request per row, poll pending work, and write each status, action, reason code, and error back without losing the source record.

Published 2026-09-11

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](https://emvero.io/bulk-email-verification/) 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](https://emvero.io/email-verification-api/) for the request contract and [security page](https://emvero.io/security/) 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.

| Output column | What you can read from it |
| --- | --- |
| `emvero_row_id` | Input fingerprint and row position |
| `emvero_submitted_email` | The trimmed copy actually sent to the API |
| `emvero_state` | Terminal result, `pending`, or an input, API, network or protocol error |
| `emvero_verification_id` | ID used to retrieve an unfinished verification |
| `emvero_status` | `deliverable`, `risky`, `undeliverable`, `unknown` or transport state `pending` |
| `emvero_action` | `allow`, `allow_with_email_confirmation` or `reject` |
| `emvero_checks` | The complete `checks` object as JSON |
| `emvero_reason_codes` | The complete `reason_codes` array as JSON |
| `emvero_confidence` | The API’s evidence score, not a delivery probability |
| `emvero_freshness` | DNS and SMTP observation times returned by the API |
| `emvero_normalized` | Domain forms from the API’s `normalized` object |
| `emvero_error_*` | Request or service errors, kept apart from the address verdict |

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](https://emvero.io/compare/real-time-vs-bulk/) 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](https://emvero.io/guides/cleaning-an-email-list/) covers sending rules. For a CRM export, the [CRM email validation guide](https://emvero.io/guides/crm-data-cleaning/) explains how to retain the customer record when its email field is unusable. The [verification result guide](https://emvero.io/blog/reading-a-verification-result/) 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](https://emvero.io/bulk-email-verification/) can help you decide how this job fits your process; it cannot promise zero bounces or successful inbox delivery.

## Terms used here

### Hard bounce and soft bounce

Common labels for permanent and temporary delivery failures; neither label alone explains the cause.

### Catch-all domain

A domain that accepts arbitrary recipient names, so an accepted address may have no individual mailbox.

### Throttling and rate limiting

Limits on connections or SMTP commands can interrupt a mailbox check without establishing whether the recipient exists.

## Situations this applies to

### Cleaning a list before a send

You are protecting the reputation of the sending domain, so the question is not which addresses are bad but which ones this send can afford.

### Cleaning a CRM without deleting good records

A CRM record is a history, not an address, so nothing here deletes anything - verification decides what you may send, never what you may keep.

[All articles](https://emvero.io/blog/)

## Choose what to check next.

Run a check from your account and inspect the evidence in its result.

Start for free