---
title: "Verificar un CSV de correo con API: guía en Python | Emvero"
description: "Verifique una lista de correo en CSV con Python y una API. Conserve las filas, retome comprobaciones pending y guarde resultados y errores por separado."
url: https://emvero.io/es/blog/validate-email-list-csv-api/
language: es
updated: 2026-10-10
---

Blog

# Cómo verificar una lista de correo desde CSV con una API

Un script lee el CSV, envía una petición de verificación por fila desde el servidor y devuelve estado, acción, códigos de motivo y errores sin perder el registro original.

Publicado 2026-09-11

Para verificar una lista de correo con una API, ejecute un proceso que lea el CSV, envíe cada dirección desde el servidor y escriba el resultado junto a la fila original. Guarde el progreso durante la ejecución: una petición puede terminar al momento, seguir en `pending` o fallar sin producir un veredicto sobre la dirección.

Esta guía incluye un script en Python para hacerlo. Usted conserva el archivo y gestiona la ejecución; Emvero recibe una dirección por petición de verificación. La página de [verificación masiva](https://emvero.io/es/bulk-email-verification/) explica este alcance del servicio. No hay un cargador de CSV alojado ni un endpoint por lotes.

## Prepare el archivo y la credencial

El CSV necesita una cabecera y una columna `email`. El ejemplo es sintético: la coma entre comillas pertenece a la nota y la última fila no tiene dirección.

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

Conserve el archivo de origen y elija otra ruta para la salida. El script mantiene las columnas y el valor original de la dirección. Solo elimina el espacio de los extremos de la copia que envía, sin convertirla a minúsculas ni reescribirla. Una dirección vacía genera un error de entrada asociado a su fila.

Hace falta Python 3, una credencial en `EMVERO_API_TOKEN` y permiso para crear la salida y el archivo de checkpoint. Elija un límite prudente de procesos de trabajo paralelos. Este ejemplo carga el CSV completo en memoria; si el archivo no cabe, adapte el proceso para leer y guardar grupos más pequeños.

Las peticiones van a `https://api.emvero.tech/v1/verifications` con `Authorization: Bearer`. Ejecute el código de forma local o en un servidor para mantener la credencial fuera del navegador. La [API de verificación de correo](https://emvero.io/es/email-verification-api/) describe la petición y la página de [seguridad](https://emvero.io/es/security/) explica el tratamiento de los datos.

## Mantenga una identidad estable para cada fila

El script calcula un hash de los bytes del archivo de entrada y del nombre de la columna de correo elegida. Ese identificador del proceso y la posición de la fila forman `emvero_row_id` y una `Idempotency-Key` estable. Si la entrada no cambia, las claves tampoco. Editar o reordenar el archivo, o elegir otra columna, crea un proceso distinto: las entradas del checkpoint anterior no se asignan a filas nuevas.

Dentro del intervalo de idempotencia admitido, repetir una clave recupera la operación original en vez de crear otra verificación. Esta guía no depende de una duración concreta del intervalo. El script también guarda el ID de cada verificación `pending`. Al reanudar, recupera ese ID antes de plantear otro `POST`.

Cada fila es un elemento independiente. Dos filas con la misma dirección reciben claves distintas; el script no elimina duplicados. Si su sistema ya tiene identificadores inmutables de registros y procesos, puede usarlos para definir su propia correspondencia.

## Escriba evidencia y errores en columnas separadas

La salida mantiene las columnas de entrada y añade las siguientes. Los objetos estructurados se guardan como JSON compacto dentro de una celda, para conservar campos anidados y arrays de códigos de motivo al escribir y volver a leer el CSV.

| Columna de salida | Qué contiene |
| --- | --- |
| `emvero_row_id` | Huella de la entrada y posición de la fila |
| `emvero_submitted_email` | Copia enviada a la API, sin espacios en los extremos |
| `emvero_state` | Resultado terminal, `pending` o error de entrada, API, red o protocolo |
| `emvero_verification_id` | ID para recuperar una verificación sin terminar |
| `emvero_status` | `deliverable`, `risky`, `undeliverable`, `unknown` o estado de transporte `pending` |
| `emvero_action` | `allow`, `allow_with_email_confirmation` o `reject` |
| `emvero_checks` | Objeto `checks` completo en JSON |
| `emvero_reason_codes` | Array `reason_codes` completo en JSON |
| `emvero_confidence` | Puntuación de la evidencia de la API, no probabilidad de entrega |
| `emvero_freshness` | Fechas de observación de DNS y SMTP devueltas por la API |
| `emvero_normalized` | Formas del dominio del objeto `normalized` |
| `emvero_error_*` | Errores de petición o servicio, separados del veredicto sobre la dirección |

Si el siguiente sistema acepta JSON o registros de base de datos, conserve los objetos en su forma nativa. Crear una columna por cada campo conocido hoy puede descartar sin aviso los que se añadan después.

## Ejecute el proceso en Python

Guarde el código como `verify_csv.py`. El parser `csv` de Python interpreta los campos entre comillas y las celdas vacías. El script rechaza las cabeceras duplicadas y los nombres reservados para resultados antes de procesar, para no sobrescribir los valores de entrada.

El checkpoint es un archivo JSON Lines al que solo se añaden eventos. Al reanudar, prevalece el último de cada fila. Los resultados terminales se reutilizan y los pendientes se recuperan por su ID. Retomar el proceso no vuelve a verificar todas las filas terminadas con evidencia nueva.

```
#!/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 rutas distintas para entrada y salida:

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

`--workers` limita cuántas filas se procesan a la vez. `--poll-seconds` establece el plazo de consulta por fila y `--checkpoint` permite elegir el archivo de progreso. Su ruta debe ser distinta de las dos rutas CSV.

El script limita los reintentos ante `429`, `500`, `503` y fallos de red. Respeta un `Retry-After` del servidor que supere la espera local. Si el error persiste, lo escribe junto a la fila. Antes de reanudar, resuelva la credencial, el saldo o el problema del servicio. Si se limitan las peticiones, reduzca la concurrencia.

## Retome la misma verificación pendiente

`POST /v1/verifications` devuelve HTTP 200 tanto para un resultado terminal como para `pending`. Guarde el `id` de la verificación pendiente, espere `retry_after_ms` y llame a `GET /v1/verifications/{id}`. Que falte un resultado terminado no justifica crear otra operación.

Si se agota el plazo de consulta del script, registra `client_poll_deadline` y mantiene `emvero_status` en `pending`. Ese código describe el límite de espera del cliente, no la dirección. Más tarde, ejecute el mismo comando con la misma entrada y el mismo checkpoint.

El contrato actual no incluye callbacks salientes. El proceso debe guardar el ID y recuperar el resultado. Una lista puede acomodar esa espera de otra forma que una petición de registro; la [comparación entre verificación en tiempo real y masiva](https://emvero.io/es/compare/real-time-vs-bulk/) explica la elección.

## Interprete un error HTTP como un problema operativo

Una respuesta HTTP 200 correcta puede contener `undeliverable`: la verificación produjo un resultado sobre la dirección. `401`, `402`, `403`, `429`, `500` y `503` son respuestas operativas. No las convierta en `undeliverable` ni invente un resultado `unknown`.

Los errores de petición aparecen en `emvero_error_http_status`, `emvero_error_code` y `emvero_error_message`. La evidencia de la dirección pertenece a `emvero_status`, `emvero_action` y los campos de resultado. Tras resolver la causa, vuelva a ejecutar el mismo proceso. El checkpoint conserva los resultados terminados y se reintentan las filas con errores.

## Aplique su política de lista a los resultados guardados

Lea `action` como recomendación para el siguiente paso, junto a la evidencia que la explica. Las recomendaciones de Emvero sirven para decisiones de registro; no autorizan el envío de campañas. Mantenga suscripciones, bajas y reglas de supresión en el sistema de envío.

- Con `deliverable`, conserve el resultado y `freshness`. La aceptación observada no garantiza la entrega futura ni la llegada a la bandeja de entrada.
- Con `risky`, lea `action` y `reason_codes` antes de elegir una regla. La evidencia de correo desechable y de catch-all puede exigir decisiones distintas. La anotación de dirección genérica por sí sola no convierte el resultado en arriesgado.
- Con `undeliverable`, excluya la dirección del envío cuando la evidencia lo respalde. Conserve la fila y el resto de los datos del cliente.
- Con `unknown`, separe el veredicto inconcluso de los grupos positivos y negativos. Puede contener evidencia útil sobre el dominio. Decida si retener o volver a comprobar la dirección. Si corresponde enviar una confirmación, debe hacerlo su propio proceso; la verificación no envía mensajes.

La [guía de limpieza de listas](https://emvero.io/es/guides/cleaning-an-email-list/) desarrolla las reglas de envío. Para una exportación del CRM, la [guía de validación del correo en CRM](https://emvero.io/es/guides/crm-data-cleaning/) explica cómo conservar el registro aunque el campo de correo no sirva. La [guía del resultado de verificación](https://emvero.io/es/blog/reading-a-verification-result/) explica los campos y por qué `confidence` no es una probabilidad.

## Revise la salida antes de entregarla

Cada fila de entrada necesita un estado registrado, también las direcciones vacías y las peticiones fallidas. Compare el número de filas y revise `input_error`, `api_error`, `network_error`, `protocol_error` y `pending` antes de pasar la salida a otro sistema.

En las filas terminales, conserve `status`, `action`, `checks`, `reason_codes`, `confidence` y `freshness` cuando la API los devuelva. Archive juntos la fuente, el identificador del proceso, la salida y el checkpoint. Mantenga el checkpoint mientras pueda necesitar reanudar; elimínelo solo cuando la salida esté guardada de forma segura.

Empiece con el pequeño CSV sintético. El [proceso de verificación masiva](https://emvero.io/es/bulk-email-verification/) ayuda a encajar este trabajo en su sistema. No puede prometer cero rebotes ni entrega en la bandeja de entrada.

## Términos que aparecen aquí

### Rebote duro y rebote blando

Dos tipos de fallo de entrega: uno permanente para el intento realizado y otro temporal que puede admitir un reintento.

### Dominio catch-all

Un dominio cuyo servidor acepta destinatarios aunque no exista un buzón para cada dirección.

### Limitación de ritmo

Un límite del servidor sobre el tráfico SMTP, que puede impedir comprobar el buzón en ese intento.

## Situaciones a las que se aplica

### Limpiar una lista antes de un envío

Está protegiendo la reputación del dominio desde el que envía, así que la pregunta no es qué direcciones son malas, sino cuáles puede permitirse este envío.

### Limpiar un CRM sin borrar registros buenos

Una ficha de CRM es una historia, no una dirección: aquí no se borra nada. La verificación decide a quién puede escribir, nunca a quién puede conservar.

[Todos los artículos](https://emvero.io/es/blog/)

## Elige qué comprobar ahora.

Haz una consulta desde tu cuenta y examina los datos del resultado.

Empezar gratis