Pythonでメールアドレスを検証する方法

import re

SYNTAX = re.compile(r'^[^\s@."]+(?:\.[^\s@."]+)*@[^\s@.]+(?:\.[^\s@.]+)+$')

def is_valid_syntax(email: str) -> bool:
    return isinstance(email, str) and len(email) <= 320 and SYNTAX.match(email) is not None

is_valid_syntax("jane@example.com")  # True
is_valid_syntax("not-an-email")      # False
is_valid_syntax("a@@b.com")          # False
is_valid_syntax("a..b@x.com")        # False

import dns.resolver  # pip install dnspython

def has_mail_route(domain: str) -> bool:
    try:
        answers = dns.resolver.resolve(domain, "MX")
        return len(answers) > 0
    except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.NoNameservers):
        return False

has_mail_route("gmail.com")              # True
has_mail_route("company-that-folded.com")  # False

import os
import requests  # pip install requests

BASE = "https://app.qualisend.com/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['QUALISEND_API_KEY']}"}

def verify(email: str) -> dict:
    res = requests.post(f"{BASE}/verify", headers=HEADERS, json={"email": email}, timeout=10)
    res.raise_for_status()
    return res.json()

import time

def verify_and_wait(email: str, tries: int = 10, delay: float = 1.5):
    data = verify(email)
    if data["result"]["smtp"] != "pending":
        return data["result"]  # cached or skipped: already final
    job_id = data["job_id"]
    for _ in range(tries):
        res = requests.get(f"{BASE}/jobs/{job_id}",
                           params={"include": "results"}, headers=HEADERS, timeout=10)
        job = res.json()
        if job["status"] == "completed":
            return job["results"][0]  # {"status": ..., "score": ..., "reason": ...}
        if job["status"] in ("canceled", "failed"):
            return None
        time.sleep(delay)
    return None  # still processing — treat as unknown, retry later

def verify_bulk(emails: list[str], tries: int = 60, delay: float = 2.0):
    res = requests.post(f"{BASE}/verify/bulk", headers=HEADERS,
                        json={"text": "\n".join(emails), "expected_count": len(emails)},
                        timeout=30)
    res.raise_for_status()
    job_id = res.json()["job_id"]

    for _ in range(tries):
        job = requests.get(f"{BASE}/jobs/{job_id}", headers=HEADERS, timeout=30).json()
        if job["status"] == "completed":
            return fetch_results(job_id)
        if job["status"] in ("canceled", "failed"):
            return None
        time.sleep(delay)  # back off; large lists take longer than a single probe
    return None

def fetch_results(job_id: str, page_size: int = 1000) -> dict:
    results, offset = {}, 0
    while True:
        page = requests.get(f"{BASE}/jobs/{job_id}", headers=HEADERS, timeout=30,
                            params={"include": "results", "limit": page_size,
                                    "offset": offset}).json()["results"]
        results.update({row["email"]: row for row in page})
        if len(page) < page_size:
            return results
        offset += page_size

def validate_email(email: str) -> dict:
    if not is_valid_syntax(email):
        return {"status": "undeliverable", "reason": "invalid_email"}
    domain = email.rsplit("@", 1)[1]
    if not has_mail_route(domain):
        return {"status": "undeliverable", "reason": "invalid_domain"}
    try:
        return verify(email)["result"]  # deliverable | risky | undeliverable | unknown
    except requests.RequestException:  # timeout, 402, 429, 5xx: fail open
        return {"status": "unknown", "reason": None}  # let it through, re-verify later