Python SDK

Python email checker SDK. pip install, verify emails with Python. Code examples and async support.

Le SDK officiel EmailVerify pour Python fournit des interfaces synchrones et asynchrones pour la vérification d'emails.

Installation

pip install emailverify

Avec Support Async

pip install emailverify[async]

Démarrage Rapide

import os
from emailverify import Client

client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])

# Vérifier un seul email
result = client.verify('user@example.com')
print(f"Status: {result.status}")  # 'valid', 'invalid', 'unknown', or 'accept_all'

Configuration

Options du Client

from emailverify import Client

client = Client(
    api_key='your-api-key',

    # Configurations optionnelles
    timeout=30,               # Délai de requête en secondes
    retries=3,                # Nombre de tentatives
    base_url='https://api.emailverify.ai',  # URL API personnalisée
)

Variables d'Environnement

# .env
EMAILVERIFY_API_KEY=your_api_key_here

import os
from emailverify import Client

client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])

Vérification d'Email Unique

Vérification Basique

result = client.verify('user@example.com')

print({
    'email': result.email,
    'status': result.status,
    'score': result.score,
    'deliverable': result.result.deliverable,
    'disposable': result.result.disposable,
    'role': result.result.role,
})

Vérification avec Options

result = client.verify(
    'user@example.com',
    smtp_check=True,      # Activer la vérification SMTP
    timeout=5,            # Délai personnalisé pour cette requête
)

Objet de Réponse

@dataclass
class VerificationResult:
    email: str
    status: str  # 'valid', 'invalid', 'unknown', 'accept_all'
    result: ResultDetails
    score: float
    reason: Optional[str]

@dataclass
class ResultDetails:
    deliverable: Optional[bool]
    valid_format: bool
    valid_domain: bool
    valid_mx: bool
    disposable: bool
    role: bool
    catchall: bool
    free: bool
    smtp_valid: Optional[bool]

Utilisation Async

Pour les applications haute performance, utilisez le client async :

import asyncio
from emailverify import AsyncClient

async def main():
    client = AsyncClient(api_key=os.environ['EMAILVERIFY_API_KEY'])

    # Vérification unique
    result = await client.verify('user@example.com')
    print(result.status)

    # Vérifier plusieurs emails de manière concurrente
    emails = ['user1@example.com', 'user2@example.com', 'user3@example.com']
    results = await asyncio.gather(*[client.verify(email) for email in emails])

    for result in results:
        print(f"{result.email}: {result.status}")

    await client.close()

asyncio.run(main())

Gestionnaire de Contexte

async with AsyncClient(api_key=os.environ['EMAILVERIFY_API_KEY']) as client:
    result = await client.verify('user@example.com')

Vérification en Masse

Soumettre une Tâche en Masse

emails = [
    'user1@example.com',
    'user2@example.com',
    'user3@example.com',
]

job = client.verify_bulk(emails)
print(f"Job ID: {job.id}")
print(f"Status: {job.status}")

Vérifier le Statut de la Tâche

status = client.get_bulk_job_status(job.id)

print({
    'id': status.id,
    'status': status.status,      # 'pending', 'processing', 'completed', 'failed'
    'total': status.total,
    'processed': status.processed,
    'valid': status.valid,
    'invalid': status.invalid,
})

Obtenir les Résultats

results = client.get_bulk_job_results(job.id)

for result in results:
    print(f"{result.email}: {result.status}")

Avec Webhook

job = client.verify_bulk(
    emails,
    webhook_url='https://your-domain.com/webhooks/emailverify',
)

Gestion des Crédits

credits = client.get_credits()

print({
    'available': credits.available,
    'used': credits.used,
    'total': credits.total,
})

Gestion des Erreurs

Types d'Exceptions

from emailverify import (
    Client,
    EmailVerifyError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
)

try:
    result = client.verify('invalid-email')
except AuthenticationError:
    print('Invalid API key')
except RateLimitError as e:
    print(f'Rate limited. Retry after {e.retry_after} seconds')
except ValidationError as e:
    print(f'Invalid input: {e}')
except EmailVerifyError as e:
    print(f'API error: {e}')

Configuration de Nouvelle Tentative

client = Client(
    api_key=os.environ['EMAILVERIFY_API_KEY'],
    retries=5,
    retry_delay=1.0,           # Initial delay in seconds
    retry_max_delay=30.0,      # Max delay between retries
)

Intégration avec les Frameworks

Django

# views.py
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from emailverify import Client
import os

client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])

@require_POST
def verify_email(request):
    import json
    data = json.loads(request.body)
    email = data.get('email')

    try:
        result = client.verify(email)

        if result.status == 'invalid':
            return JsonResponse({
                'valid': False,
                'message': 'Please enter a valid email address',
            }, status=400)

        if result.result.disposable:
            return JsonResponse({
                'valid': False,
                'message': 'Disposable emails are not allowed',
            }, status=400)

        return JsonResponse({'valid': True, 'score': result.score})
    except Exception as e:
        return JsonResponse({'error': str(e)}, status=500)

Validation de Formulaire Django

# forms.py
from django import forms
from emailverify import Client
import os

client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])

class RegistrationForm(forms.Form):
    email = forms.EmailField()

    def clean_email(self):
        email = self.cleaned_data['email']
        result = client.verify(email)

        if result.status == 'invalid':
            raise forms.ValidationError('Please enter a valid email address')

        if result.result.disposable:
            raise forms.ValidationError('Disposable emails are not allowed')

        return email

Flask

from flask import Flask, request, jsonify
from emailverify import Client
import os

app = Flask(__name__)
client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])

@app.route('/verify', methods=['POST'])
def verify_email():
    email = request.json.get('email')

    try:
        result = client.verify(email)
        return jsonify({
            'email': result.email,
            'status': result.status,
            'score': result.score,
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500

FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from emailverify import AsyncClient
import os

app = FastAPI()
client = AsyncClient(api_key=os.environ['EMAILVERIFY_API_KEY'])

class EmailRequest(BaseModel):
    email: str

@app.post('/verify')
async def verify_email(request: EmailRequest):
    try:
        result = await client.verify(request.email)
        return {
            'email': result.email,
            'status': result.status,
            'score': result.score,
            'deliverable': result.result.deliverable,
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.on_event('shutdown')
async def shutdown():
    await client.close()

Exemples Complets

Validation d'Inscription Utilisateur

from emailverify import Client
import os

client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])

def validate_registration_email(email: str) -> dict:
    """Valider l'email pour l'inscription utilisateur."""
    result = client.verify(email)

    # Reject invalid emails
    if result.status == 'invalid':
        return {
            'valid': False,
            'reason': 'invalid_email',
            'message': 'This email address is invalid',
        }

    # Reject disposable emails
    if result.result.disposable:
        return {
            'valid': False,
            'reason': 'disposable',
            'message': 'Please use a permanent email address',
        }

    # Warn about role-based emails
    if result.result.role:
        return {
            'valid': True,
            'warning': 'role_based',
            'message': 'Using a personal email is recommended',
        }

    return {
        'valid': True,
        'score': result.score,
    }

# Utilisation
validation = validate_registration_email('user@example.com')
if not validation['valid']:
    print(validation['message'])

Nettoyage de Liste en Masse

import csv
from emailverify import Client
import os

client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])

def clean_email_list(input_file: str, output_file: str):
    """Nettoyer une liste d'emails et exporter les résultats."""

    # Lire les emails depuis CSV
    with open(input_file, 'r') as f:
        reader = csv.DictReader(f)
        emails = [row['email'] for row in reader]

    # Soumettre vérification en masse
    job = client.verify_bulk(emails)
    print(f"Submitted job {job.id} with {len(emails)} emails")

    # Attendre la fin (interrogation)
    import time
    while True:
        status = client.get_bulk_job_status(job.id)
        print(f"Progress: {status.processed}/{status.total}")

        if status.status == 'completed':
            break
        elif status.status == 'failed':
            raise Exception(f"Job failed: {status.error}")

        time.sleep(5)

    # Obtenir les résultats et écrire vers la sortie
    results = client.get_bulk_job_results(job.id)

    with open(output_file, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=['email', 'status', 'score', 'disposable'])
        writer.writeheader()

        for result in results:
            writer.writerow({
                'email': result.email,
                'status': result.status,
                'score': result.score,
                'disposable': result.result.disposable,
            })

    print(f"Results written to {output_file}")
    print(f"Valid: {status.valid}, Invalid: {status.invalid}")

# Utilisation
clean_email_list('contacts.csv', 'cleaned_contacts.csv')

Vérification Concurrente Async

import asyncio
from emailverify import AsyncClient
import os

async def verify_emails_concurrent(emails: list[str], concurrency: int = 10):
    """Vérifier plusieurs emails avec concurrence contrôlée."""

    client = AsyncClient(api_key=os.environ['EMAILVERIFY_API_KEY'])
    semaphore = asyncio.Semaphore(concurrency)

    async def verify_with_semaphore(email):
        async with semaphore:
            return await client.verify(email)

    tasks = [verify_with_semaphore(email) for email in emails]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    await client.close()

    # Process results
    valid = []
    invalid = []
    errors = []

    for email, result in zip(emails, results):
        if isinstance(result, Exception):
            errors.append((email, str(result)))
        elif result.status == 'valid':
            valid.append(email)
        else:
            invalid.append(email)

    return {
        'valid': valid,
        'invalid': invalid,
        'errors': errors,
    }

# Utilisation
emails = ['user1@example.com', 'user2@example.com', 'user3@example.com']
results = asyncio.run(verify_emails_concurrent(emails))
print(f"Valid: {len(results['valid'])}, Invalid: {len(results['invalid'])}")

Annotations de Type

Le SDK inclut des annotations de type complètes pour le support IDE :

from emailverify import Client, VerificationResult, BulkJob
from typing import Optional

def process_verification(result: VerificationResult) -> Optional[str]:
    if result.status == 'valid':
        return result.email
    return None