Automating Email Hygiene via API: Architecture and Code

Manual list cleaning works fine until your database grows past a thousand addresses. After that it becomes a routine: export CSV, upload to a service, wait, download, re-import. Same steps every time. An API lets you cut the human out of that loop entirely and build a pipeline that cleans your list on a schedule, with no manual work.

Why manual hygiene does not scale

An email list degrades constantly. People change jobs, ISPs disable inactive mailboxes, domains expire. Across various lists, 2-3% of addresses go invalid every month. For a list of 50,000 contacts, that is 1,000 to 1,500 new dead addresses each month.

A marketer who remembers to clean the list once a quarter finds 5-8% invalid addresses by the time they get around to it. At that point the bounce rate has already dented domain reputation, the ESP has started throttling sends, and some messages are landing in spam even for subscribers whose addresses are perfectly valid.

Automation solves this: a script runs on a cron schedule, submits addresses to the validation API, retrieves results, and updates the database. No CSV, no web interface, no human in the loop.

Pipeline architecture

A typical automated hygiene pipeline has four stages: extract addresses from the database or CRM; submit them for validation via the API; retrieve results; update statuses in the database.

There are subtleties between stages. Not every address needs checking on every run — only those not validated recently or that bounced on the last send. Results must be interpreted carefully: "risky" is not the same as "invalid". And the database update should be idempotent — running the script twice must not corrupt data.

The sections below cover each stage with code in both Python and Node.js.

Step 1. Selecting addresses to validate

Checking the entire list every run is wasteful. An address validated three days ago and marked good almost certainly has not changed. The practical strategy: validate addresses that have not been checked for more than N days, plus any that bounced on the last send.

# Python: fetch emails not validated in 30+ days
import psycopg2
from datetime import datetime, timedelta

def get_stale_emails(conn, days: int = 30, limit: int = 50_000) -> list[str]:
    """Fetch emails that haven't been validated recently."""
    cutoff = datetime.utcnow() - timedelta(days=days)
    with conn.cursor() as cur:
        cur.execute("""
            SELECT email FROM subscribers
            WHERE status = 'active'
              AND (last_validated_at IS NULL OR last_validated_at < %s)
            ORDER BY last_validated_at ASC NULLS FIRST
            LIMIT %s
        """, (cutoff, limit))
        return [row[0] for row in cur.fetchall()]

The same query in Node.js with PostgreSQL:

// Node.js: select stale emails for validation
const { Pool } = require("pg");
const pool = new Pool();

async function getStaleEmails(days = 30, limit = 50000) {
  const cutoff = new Date(Date.now() - days * 86400000);
  const { rows } = await pool.query(
    `SELECT email FROM subscribers
     WHERE status = 'active'
       AND (last_validated_at IS NULL OR last_validated_at < $1)
     ORDER BY last_validated_at ASC NULLS FIRST
     LIMIT $2`,
    [cutoff, limit]
  );
  return rows.map((r) => r.email);
}

A limit of 50,000 addresses per run is a reasonable ceiling. If the list is larger, the next run picks up the next batch. At that pace a daily cron will cycle through 350K addresses in a week.

The NULLS FIRST ordering ensures addresses that have never been validated at all go to the front of the queue, before those that have been checked at least once.

Step 2. Submitting for validation

Validation APIs typically expose two endpoints: single (one address per request) and bulk (an array of addresses). For automated hygiene, bulk is the only sensible choice — one HTTP request instead of fifty thousand.

The bulk endpoint accepts an array and returns a task_id. Validation runs asynchronously; results are fetched separately. Most services cap batch size at 50,000 addresses, so larger lists need splitting.

# Python: submit batches for validation
import os
import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

API_KEY = os.environ["UCHECKER_API_KEY"]
BASE = "https://api.uchecker.net"
BATCH_SIZE = 10_000

session = requests.Session()
retry = Retry(total=3, backoff_factor=2, status_forcelist=[429, 500, 502, 503])
session.mount("https://", HTTPAdapter(max_retries=retry))

def chunk(lst: list, size: int):
    for i in range(0, len(lst), size):
        yield lst[i : i + size]

def submit_for_validation(emails: list[str]) -> list[int]:
    """Submit emails in batches, return list of task IDs."""
    task_ids = []
    batches = list(chunk(emails, BATCH_SIZE))

    for i, batch in enumerate(batches):
        resp = session.post(
            f"{BASE}/api/v1/validate/bulk",
            headers={"x-api-key": API_KEY},
            json={"emails": batch},
        )
        resp.raise_for_status()
        data = resp.json()
        task_ids.append(data["task_id"])
        print(f"Batch {i+1}/{len(batches)}: task_id={data['task_id']}, "
              f"queued={data['valid_emails']}")
        time.sleep(1)  # avoid rate limits between batches

    return task_ids

Node.js version:

// Node.js: submit batches for validation
const API_KEY = process.env.UCHECKER_API_KEY;
const BASE = "https://api.uchecker.net";
const BATCH_SIZE = 10_000;

function chunk(arr, size) {
  const result = [];
  for (let i = 0; i < arr.length; i += size) {
    result.push(arr.slice(i, i + size));
  }
  return result;
}

async function submitForValidation(emails) {
  const batches = chunk(emails, BATCH_SIZE);
  const taskIds = [];

  for (let i = 0; i < batches.length; i++) {
    const resp = await fetch(`${BASE}/api/v1/validate/bulk`, {
      method: "POST",
      headers: {
        "x-api-key": API_KEY,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({ emails: batches[i] }),
    });

    if (!resp.ok) throw new Error(`HTTP ${resp.status}: ${await resp.text()}`);
    const data = await resp.json();
    taskIds.push(data.task_id);
    console.log(`Batch ${i + 1}/${batches.length}: task_id=${data.task_id}`);

    if (i < batches.length - 1) {
      await new Promise((r) => setTimeout(r, 1000));
    }
  }
  return taskIds;
}

HTTP-level retry logic is not optional. Networks fail, servers return 500 under load. Without retries a script can fall on batch three of twenty, and you discover this the next morning from cron logs. Python handles this via urllib3 Retry; in Node.js use a retry wrapper or a library like p-retry.

Step 3. Retrieving results

After submission the server queues the task. Validating 10,000 addresses takes 3-5 minutes. Results come back via polling or webhook.

Polling is simpler to implement: query the task endpoint until status becomes completed. With exponential backoff polling puts negligible load on the API — start at 5 seconds, grow the interval up to a minute.

# Python: poll task with exponential backoff
def poll_task(task_id: int, timeout: int = 900) -> dict:
    """Wait for task completion. Returns results on success."""
    interval = 5
    max_interval = 60
    elapsed = 0

    while elapsed < timeout:
        resp = session.get(
            f"{BASE}/api/v1/tasks/{task_id}",
            headers={"x-api-key": API_KEY},
        )
        data = resp.json()

        if data["status"] == "completed":
            return data
        if data["status"] == "failed":
            raise RuntimeError(f"Task {task_id} failed: {data.get('error')}")

        time.sleep(interval)
        elapsed += interval
        interval = min(interval * 1.5, max_interval)

    raise TimeoutError(f"Task {task_id} timed out after {timeout}s")

// Node.js: poll task
async function pollTask(taskId, timeoutMs = 900000) {
  let interval = 5000;
  const maxInterval = 60000;
  const start = Date.now();

  while (Date.now() - start < timeoutMs) {
    const resp = await fetch(`${BASE}/api/v1/tasks/${taskId}`, {
      headers: { "x-api-key": API_KEY },
    });
    const data = await resp.json();

    if (data.status === "completed") return data;
    if (data.status === "failed") {
      throw new Error(`Task ${taskId} failed: ${data.error}`);
    }

    await new Promise((r) => setTimeout(r, interval));
    interval = Math.min(interval * 1.5, maxInterval);
  }
  throw new Error(`Task ${taskId} timed out`);
}

Once a task completes, fetch the detailed results:

# Python: fetch results for a completed task
def fetch_results(task_id: int) -> list[dict]:
    """Get detailed validation results for a completed task."""
    resp = session.get(
        f"{BASE}/api/v1/tasks/{task_id}/results",
        headers={"x-api-key": API_KEY},
        params={"format": "json"},
    )
    resp.raise_for_status()
    return resp.json()["data"]
    # Each item: {"email": "...", "validation_result": "good"|"bad"|"risky",
    #             "reason": "mailbox_not_found"|"disposable"|...}

When you have multiple tasks, poll them in parallel. In Python, use concurrent.futures.ThreadPoolExecutor; in Node.js, use Promise.allSettled. Polling five tasks sequentially at 5 minutes each takes 25 minutes; in parallel it takes 5.

// Node.js: poll multiple tasks in parallel
async function pollAllTasks(taskIds) {
  const results = await Promise.allSettled(
    taskIds.map((id) => pollTask(id))
  );
  const completed = [];
  for (const [i, result] of results.entries()) {
    if (result.status === "fulfilled") {
      completed.push(taskIds[i]);
    } else {
      console.error(`Task ${taskIds[i]}: ${result.reason.message}`);
    }
  }
  return completed;
}

Step 4. Updating the database

Writing results back to the database is more than tagging an address as valid or invalid. Store the reason code and the validation timestamp too — that lets you filter by problem type and track trends over time.

# Python: write validation results to the database
from datetime import datetime

def update_subscriber_statuses(conn, results: list[dict]):
    """Update subscribers based on validation results."""
    now = datetime.utcnow()
    good, bad, risky = 0, 0, 0

    with conn.cursor() as cur:
        for r in results:
            email = r["email"]
            verdict = r["validation_result"]  # "good", "bad", "risky"
            reason = r.get("reason", "")

            if verdict == "bad":
                cur.execute("""
                    UPDATE subscribers
                    SET status = 'invalid', validation_reason = %s,
                        last_validated_at = %s
                    WHERE email = %s
                """, (reason, now, email))
                bad += 1
            elif verdict == "risky":
                cur.execute("""
                    UPDATE subscribers
                    SET status = 'risky', validation_reason = %s,
                        last_validated_at = %s
                    WHERE email = %s
                """, (reason, now, email))
                risky += 1
            else:
                cur.execute("""
                    UPDATE subscribers
                    SET status = 'active', validation_reason = NULL,
                        last_validated_at = %s
                    WHERE email = %s
                """, (now, email))
                good += 1

    conn.commit()
    print(f"Updated: {good} good, {bad} bad, {risky} risky")
    return {"good": good, "bad": bad, "risky": risky}

Row-by-row UPDATE is slow at scale. For large result sets, group addresses by verdict and batch-update each group:

# Python: batch update via executemany
def update_batch(conn, results: list[dict]):
    """Batch update - faster for large result sets."""
    now = datetime.utcnow()
    bad_emails = [(r["reason"], now, r["email"])
                  for r in results if r["validation_result"] == "bad"]
    risky_emails = [(r["reason"], now, r["email"])
                    for r in results if r["validation_result"] == "risky"]
    good_emails = [(now, r["email"])
                   for r in results if r["validation_result"] == "good"]

    with conn.cursor() as cur:
        if bad_emails:
            cur.executemany("""
                UPDATE subscribers
                SET status='invalid', validation_reason=%s, last_validated_at=%s
                WHERE email=%s
            """, bad_emails)
        if risky_emails:
            cur.executemany("""
                UPDATE subscribers
                SET status='risky', validation_reason=%s, last_validated_at=%s
                WHERE email=%s
            """, risky_emails)
        if good_emails:
            cur.executemany("""
                UPDATE subscribers
                SET status='active', validation_reason=NULL, last_validated_at=%s
                WHERE email=%s
            """, good_emails)

    conn.commit()
    print(f"Batch update: {len(good_emails)} good, "
          f"{len(bad_emails)} bad, {len(risky_emails)} risky")

The risky segment deserves its own decision. Catch-all domains and disposable addresses end up here — technically reachable, but higher risk. Deleting them outright means losing real subscribers. Keeping them without differentiation invites bounces. A workable middle ground: move risky addresses into a separate segment and mail them less frequently. If one bounces, promote it to invalid.

Putting it all together

Full script — run manually or via cron:

#!/usr/bin/env python3
"""Automated email hygiene via uChecker API.

Run daily via cron:
  0 3 * * * /usr/bin/python3 /opt/scripts/email_hygiene.py >> /var/log/email_hygiene.log 2>&1
"""
import os
import sys
import psycopg2
from datetime import datetime

# ... functions from above: get_stale_emails, submit_for_validation,
#     poll_task, fetch_results, update_batch

def run():
    conn = psycopg2.connect(os.environ["DATABASE_URL"])

    # 1. Get emails that need checking
    emails = get_stale_emails(conn, days=30, limit=50_000)
    if not emails:
        print(f"{datetime.utcnow()}: No stale emails. Exiting.")
        return

    print(f"{datetime.utcnow()}: Found {len(emails)} emails to validate")

    # 2. Check credit balance
    balance = session.get(
        f"{BASE}/api/v1/account/balance",
        headers={"x-api-key": API_KEY},
    ).json()["credits_remaining"]

    if balance < len(emails):
        print(f"Insufficient credits: {balance} < {len(emails)}")
        sys.exit(1)

    # 3. Submit for validation
    task_ids = submit_for_validation(emails)

    # 4. Wait for results
    all_results = []
    for tid in task_ids:
        poll_task(tid)
        results = fetch_results(tid)
        all_results.extend(results)

    # 5. Update database
    update_batch(conn, all_results)

    conn.close()
    print(f"{datetime.utcnow()}: Done. Processed {len(all_results)} emails.")

if __name__ == "__main__":
    run()

Node.js equivalent for JavaScript backends:

// email-hygiene.js - Node.js version
const { Pool } = require("pg");
const pool = new Pool();

async function run() {
  const emails = await getStaleEmails(30, 50000);
  if (!emails.length) {
    console.log("No stale emails. Exiting.");
    return;
  }
  console.log(`Found ${emails.length} emails to validate`);

  const taskIds = await submitForValidation(emails);
  const completedIds = await pollAllTasks(taskIds);

  for (const tid of completedIds) {
    const results = await fetchResults(tid);
    await updateSubscribers(pool, results);
  }

  console.log("Done.");
  await pool.end();
}

run().catch((err) => {
  console.error(err);
  process.exit(1);
});

Cron setup and monitoring

Running at 3 AM daily is the standard choice. Database load is low, the API is not saturated by daytime traffic, and results are ready before the morning send.

# crontab -e
0 3 * * * /usr/bin/python3 /opt/scripts/email_hygiene.py >> /var/log/email_hygiene.log 2>&1

# Or weekly if the list is small
0 3 * * 1 /usr/bin/python3 /opt/scripts/email_hygiene.py >> /var/log/email_hygiene.log 2>&1

A script that fails silently is worse than no script at all — it creates a false sense of a working system. Add failure alerts. The simplest approach sends a Slack notification on any non-zero exit code:

# Cron wrapper with Slack alert on failure
#!/bin/bash
SLACK_WEBHOOK="https://hooks.slack.com/services/YOUR/WEBHOOK/URL"

/usr/bin/python3 /opt/scripts/email_hygiene.py >> /var/log/email_hygiene.log 2>&1
EXIT_CODE=$?

if [ $EXIT_CODE -ne 0 ]; then
  curl -s -X POST "$SLACK_WEBHOOK" \
    -H "Content-Type: application/json" \
    -d "{\"text\":\"Email hygiene script failed with exit code $EXIT_CODE\"}"
fi

For proper observability, emit metrics after every run: addresses checked, how many were invalid, credits spent, total runtime. If the invalid rate jumps from 3% to 15%, that is a signal — either address collection is broken somewhere, or a specific provider just mass-deactivated mailboxes.

Webhooks as an alternative to polling

Polling is convenient for cron scripts that block and wait for results. If you have a web server running, webhooks are more efficient: the API sends a POST to your endpoint when the task finishes. No process to keep alive, no status queries.

# Submit a batch with a webhook URL
resp = session.post(
    f"{BASE}/api/v1/validate/bulk",
    headers={"x-api-key": API_KEY},
    json={
        "emails": batch,
        "webhook_url": "https://your-app.com/webhooks/email-validation",
    },
)

// Express: handle incoming webhook
const express = require("express");
const app = express();

app.post("/webhooks/email-validation", express.json(), async (req, res) => {
  const { task_id, status } = req.body;

  if (status === "completed") {
    const results = await fetchResults(task_id);
    await updateSubscribers(pool, results);
    console.log(`Webhook: task ${task_id} processed, ${results.length} emails`);
  }

  res.sendStatus(200);
});

In production, combine both. Webhook is the primary channel. A polling loop with a long interval (once a minute) acts as a fallback for cases where the webhook missed delivery due to a network blip or a server restart.

Validation at the point of entry

The cron pipeline cleans existing records. But bad addresses keep arriving through subscription forms, CRM imports, and integrations right now. If you only validate them after 30 days, you spend a month sending to dead mailboxes.

The single-address endpoint handles this. One call, one address, response in under a second. Drop it into the signup handler or the import processor.

# Python: validate a single address at signup
def validate_on_signup(email: str) -> dict:
    """Validate a single email at signup. Returns verdict."""
    resp = session.post(
        f"{BASE}/api/v1/validate/single",
        headers={"x-api-key": API_KEY},
        json={"email": email},
        timeout=5,
    )
    resp.raise_for_status()
    data = resp.json()
    return {
        "valid": data["result"] != "invalid",
        "reason": data.get("reason"),
        "risk_score": data.get("risk_score", 0),
    }

// Node.js: middleware for email validation at signup
async function validateEmail(email) {
  const resp = await fetch(`${BASE}/api/v1/validate/single`, {
    method: "POST",
    headers: {
      "x-api-key": API_KEY,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ email }),
  });
  const data = await resp.json();
  return {
    valid: data.result !== "invalid",
    reason: data.reason ?? null,
    riskScore: data.risk_score ?? 0,
  };
}

// Usage in Express route:
app.post("/subscribe", async (req, res) => {
  const { email } = req.body;
  const check = await validateEmail(email);

  if (!check.valid) {
    return res.status(400).json({ error: "Invalid email address" });
  }
  // ... save subscriber
});

Two layers together — validation at entry and periodic re-validation — cover hygiene end to end. The entry check keeps garbage out from the start. The cron catches addresses that went bad after a subscriber joined.

Common mistakes

Re-validating the entire list every run. An address that was good a week ago almost certainly still is. Spending a credit on it again is wasteful. A last_validated_at field solves this — only process stale addresses.

Deleting risky addresses without review. Catch-all domains land in risky, but real subscribers sit behind them. Move risky to a separate segment rather than deleting immediately.

No checkpoints. The script submits five batches, crashes on the sixth. On restart it resubmits everything — double credit spend. Save each task_id to a file or database row right after submission.

Ignoring rate limits. Ten POST requests per second and the API starts returning 429. Pauses between batches and retry with backoff prevent this before it becomes a problem.

No failure alerts. A cron job that fails silently creates a false sense of a running system. Alert on any non-zero exit code.

Automated hygiene is not a one-time setup. It is a pipeline that runs every day, at any volume, through any failure. The gap between a script and a pipeline is error handling, checkpoints, and monitoring.