Validate Emails in Python with the Syvel API

import httpx
import os
import sys

API_KEY = os.environ.get("SYVEL_API_KEY")
BASE_URL = "https://api.syvel.io/v1"

def validate_email(email: str) -> dict:
    """Validates an email via the Syvel API. Returns the complete result."""
    with httpx.Client(timeout=10.0) as client:
        response = client.get(
            f"{BASE_URL}/check/{email}",
            headers={"Authorization": f"Bearer {API_KEY}"},
        )
        response.raise_for_status()
        return response.json()

def main():
    email = sys.argv[1] if len(sys.argv) > 1 else input("Email to validate: ")
    result = validate_email(email)

    print(f"\n📧 Analysis of: {email}")
    print(f"   Risk score        : {result['risk_score']}/100")
    print(f"   Deliverability    : {result['deliverability_score']}/100")
    print(f"   Risky             : {'⚠ Yes' if result['is_risky'] else '✓ No'}")
    print(f"   Reason            : {result['reason']}")
    print(f"   MX Provider       : {result.get('mx_provider_label') or 'Unknown'}")

    if result.get('did_you_mean'):
        print(f"   Did you mean      : {result['did_you_mean']}")

if __name__ == "__main__":
    main()

import httpx
import os
import logging
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel, EmailStr

router = APIRouter()
logger = logging.getLogger(__name__)

SYVEL_API_KEY = os.environ.get("SYVEL_API_KEY")
SYVEL_BASE_URL = "https://api.syvel.io/v1/check"

class SyvelResult(BaseModel):
    risk_score: int
    is_risky: bool
    reason: str  # "safe" | "disposable" | "undeliverable" | "role_account"
    did_you_mean: str | None = None
    deliverability_score: int

async def check_email_quality(email: str, client: httpx.AsyncClient) -> SyvelResult | None:
    """
    Checks email quality via Syvel.
    Returns None on error (fail open — never block on external error).
    """
    if not SYVEL_API_KEY:
        logger.warning("SYVEL_API_KEY not configured, validation skipped")
        return None
    try:
        response = await client.get(
            f"{SYVEL_BASE_URL}/{email}",
            headers={"Authorization": f"Bearer {SYVEL_API_KEY}"},
            timeout=5.0,
        )
        if response.status_code == 200:
            return SyvelResult(**response.json())
    except httpx.TimeoutException:
        logger.warning(f"Syvel timeout for {email}")
    except Exception as e:
        logger.error(f"Syvel unexpected error: {e}")
    return None


class RegisterRequest(BaseModel):
    email: EmailStr
    password: str
    name: str

@router.post("/register", status_code=status.HTTP_201_CREATED)
async def register(body: RegisterRequest):
    async with httpx.AsyncClient() as client:
        quality = await check_email_quality(body.email, client)

    if quality is not None:
        # Block risky emails (disposable, undeliverable)
        if quality.is_risky:
            detail = {
                "field": "email",
                "code": f"email_{quality.reason}",
            }
            if quality.reason == "undeliverable":
                detail["message"] = "This domain cannot receive emails."
            elif quality.reason == "role_account":
                detail["message"] = "Please use a personal email address (not admin@, info@…)."
            else:
                detail["message"] = "This email address is not accepted."

            # Suggest a correction if typo detected
            if quality.did_you_mean:
                detail["did_you_mean"] = quality.did_you_mean

            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail=detail,
            )

    # Here: create user in database, hash password, etc.
    return {"status": "created", "email": body.email}

from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI

http_client: httpx.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global http_client
    http_client = httpx.AsyncClient(timeout=5.0)
    yield
    await http_client.aclose()

app = FastAPI(lifespan=lifespan)

# In your router, inject via dependency:
from fastapi import Depends

def get_http_client() -> httpx.AsyncClient:
    return http_client

@router.post("/register")
async def register(body: RegisterRequest, client: httpx.AsyncClient = Depends(get_http_client)):
    quality = await check_email_quality(body.email, client)
    ...

import httpx
import os
import logging
from django.core.exceptions import ValidationError

logger = logging.getLogger(__name__)

SYVEL_API_KEY = os.environ.get("SYVEL_API_KEY")
SYVEL_BASE_URL = "https://api.syvel.io/v1/check"


def validate_email_quality(email: str) -> None:
    """
    Django validator — raises ValidationError if the email is unacceptable.
    Fail open on external error.
    """
    if not SYVEL_API_KEY:
        return  # Validation disabled if key not configured

    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.get(
                f"{SYVEL_BASE_URL}/{email}",
                headers={"Authorization": f"Bearer {SYVEL_API_KEY}"},
            )
            if response.status_code != 200:
                return  # Fail open
            data = response.json()
    except Exception as e:
        logger.error(f"Syvel validation error for {email}: {e}")
        return  # Fail open

    if data.get("is_risky"):
        reason = data.get("reason", "disposable")
        if reason == "undeliverable":
            raise ValidationError(
                "This domain cannot receive emails.",
                code="email_undeliverable",
            )
        elif reason == "role_account":
            raise ValidationError(
                "Please use a personal email address.",
                code="email_role_account",
            )
        else:
            raise ValidationError(
                "This email address is not accepted. Please use your professional email.",
                code="email_disposable",
            )

from django.db import models
from django.core.validators import validate_email
from .validators import validate_email_quality

class UserProfile(models.Model):
    email = models.EmailField(
        unique=True,
        validators=[validate_email_quality],  # Applies to full_clean() and forms
    )
    name = models.CharField(max_length=255)
    created_at = models.DateTimeField(auto_now_add=True)

from django import forms
from .validators import validate_email_quality

class RegistrationForm(forms.Form):
    name = forms.CharField(max_length=255)
    email = forms.EmailField(validators=[validate_email_quality])
    password = forms.CharField(widget=forms.PasswordInput)

import asyncio
import csv
import httpx
import os
import logging
from dataclasses import dataclass, asdict
from pathlib import Path

logger = logging.getLogger(__name__)

SYVEL_API_KEY = os.environ.get("SYVEL_API_KEY")
CONCURRENCY = 10  # Number of simultaneous requests

@dataclass
class ValidationResult:
    email: str
    risk_score: int
    is_risky: bool
    reason: str
    deliverability_score: int
    did_you_mean: str
    error: str = ""

async def validate_one(email: str, client: httpx.AsyncClient, semaphore: asyncio.Semaphore) -> ValidationResult:
    async with semaphore:
        try:
            r = await client.get(
                f"https://api.syvel.io/v1/check/{email}",
                headers={"Authorization": f"Bearer {SYVEL_API_KEY}"},
                timeout=10.0,
            )
            if r.status_code == 200:
                d = r.json()
                return ValidationResult(
                    email=email,
                    risk_score=d["risk_score"],
                    is_risky=d["is_risky"],
                    reason=d["reason"],
                    deliverability_score=d["deliverability_score"],
                    did_you_mean=d.get("did_you_mean") or "",
                )
            elif r.status_code == 429:
                # Rate limit — wait and retry
                await asyncio.sleep(2)
                return await validate_one(email, client, semaphore)
        except Exception as e:
            logger.error(f"Error validating {email}: {e}")
        return ValidationResult(
            email=email, risk_score=-1, is_risky=False,
            reason="error", deliverability_score=-1,
            did_you_mean="", error="API error",
        )

async def validate_batch(emails: list[str]) -> list[ValidationResult]:
    semaphore = asyncio.Semaphore(CONCURRENCY)
    async with httpx.AsyncClient() as client:
        tasks = [validate_one(email, client, semaphore) for email in emails]
        results = await asyncio.gather(*tasks)
    return list(results)

def main():
    input_file = Path("emails.csv")
    output_valid = Path("emails_valid.csv")
    output_blocked = Path("emails_blocked.csv")
    output_errors = Path("emails_errors.csv")

    with open(input_file) as f:
        reader = csv.DictReader(f)
        emails = [row["email"].strip() for row in reader if row.get("email")]

    print(f"🔍 Validating {len(emails)} addresses...")
    results = asyncio.run(validate_batch(emails))

    valid   = [r for r in results if not r.is_risky and r.risk_score >= 0]
    blocked = [r for r in results if r.is_risky]
    errors  = [r for r in results if r.risk_score < 0]

    # Write results
    fields = ["email", "risk_score", "is_risky", "reason", "deliverability_score", "did_you_mean", "error"]
    for path, data in [(output_valid, valid), (output_blocked, blocked), (output_errors, errors)]:
        with open(path, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=fields)
            writer.writeheader()
            writer.writerows([asdict(r) for r in data])

    print(f"✅ Valid: {len(valid)}")
    print(f"🚫 Blocked: {len(blocked)}")
    print(f"⚠️  Errors: {len(errors)}")
    print(f"\nGenerated files: {output_valid}, {output_blocked}, {output_errors}")

if __name__ == "__main__":
    main()