Python SDK

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

Das offizielle EmailVerify SDK für Python bietet sowohl synchrone als auch asynchrone Schnittstellen für die E-Mail-Verifizierung.

Installation

pip install emailverify

Mit Async-Unterstützung

pip install emailverify[async]

Schnellstart

import os
from emailverify import Client

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

# Eine einzelne E-Mail verifizieren
result = client.verify('user@example.com')
print(f"Status: {result.status}")  # 'valid', 'invalid', 'unknown' oder 'accept_all'

Konfiguration

Client-Optionen

from emailverify import Client

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

    # Optionale Konfigurationen
    timeout=30,               # Anfrage-Timeout in Sekunden
    retries=3,                # Anzahl der Wiederholungsversuche
    base_url='https://api.emailverify.ai',  # Benutzerdefinierte API-URL
)

Umgebungsvariablen

# .env
EMAILVERIFY_API_KEY=your_api_key_here

import os
from emailverify import Client

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

Einzelne E-Mail-Verifizierung

Grundlegende Verifizierung

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,
})

Verifizierung mit Optionen

result = client.verify(
    'user@example.com',
    smtp_check=True,      # SMTP-Verifizierung aktivieren
    timeout=5,            # Benutzerdefinierter Timeout für diese Anfrage
)

Antwortobjekt

@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]

Async-Verwendung

Für leistungsstarke Anwendungen verwenden Sie den Async-Client:

import asyncio
from emailverify import AsyncClient

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

    # Einzelverifizierung
    result = await client.verify('user@example.com')
    print(result.status)

    # Mehrere E-Mails gleichzeitig verifizieren
    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())

Context Manager

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

Massenverifizierung

Massenauftrag einreichen

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}")

Auftragsstatus prüfen

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,
})

Ergebnisse abrufen

results = client.get_bulk_job_results(job.id)

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

Mit Webhook

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

Guthaben-Verwaltung

credits = client.get_credits()

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

Fehlerbehandlung

Ausnahmetypen

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

try:
    result = client.verify('invalid-email')
except AuthenticationError:
    print('Ungültiger API-Schlüssel')
except RateLimitError as e:
    print(f'Rate-Limit erreicht. Wiederholen nach {e.retry_after} Sekunden')
except ValidationError as e:
    print(f'Ungültige Eingabe: {e}')
except EmailVerifyError as e:
    print(f'API-Fehler: {e}')

Wiederholungskonfiguration

client = Client(
    api_key=os.environ['EMAILVERIFY_API_KEY'],
    retries=5,
    retry_delay=1.0,           # Anfängliche Verzögerung in Sekunden
    retry_max_delay=30.0,      # Maximale Verzögerung zwischen Wiederholungen
)

Framework-Integration

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': 'Bitte geben Sie eine gültige E-Mail-Adresse ein',
            }, status=400)

        if result.result.disposable:
            return JsonResponse({
                'valid': False,
                'message': 'Wegwerf-E-Mails sind nicht erlaubt',
            }, status=400)

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

Django-Formularvalidierung

# 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('Bitte geben Sie eine gültige E-Mail-Adresse ein')

        if result.result.disposable:
            raise forms.ValidationError('Wegwerf-E-Mails sind nicht erlaubt')

        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()

Vollständige Beispiele

Benutzerregistrierungsvalidierung

from emailverify import Client
import os

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

def validate_registration_email(email: str) -> dict:
    """E-Mail für Benutzerregistrierung validieren."""
    result = client.verify(email)

    # Ungültige E-Mails ablehnen
    if result.status == 'invalid':
        return {
            'valid': False,
            'reason': 'invalid_email',
            'message': 'Diese E-Mail-Adresse ist ungültig',
        }

    # Wegwerf-E-Mails ablehnen
    if result.result.disposable:
        return {
            'valid': False,
            'reason': 'disposable',
            'message': 'Bitte verwenden Sie eine permanente E-Mail-Adresse',
        }

    # Warnung bei rollenbasierten E-Mails
    if result.result.role:
        return {
            'valid': True,
            'warning': 'role_based',
            'message': 'Die Verwendung einer persönlichen E-Mail wird empfohlen',
        }

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

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

Massenlistenbereinigung

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):
    """E-Mail-Liste bereinigen und Ergebnisse exportieren."""

    # E-Mails aus CSV lesen
    with open(input_file, 'r') as f:
        reader = csv.DictReader(f)
        emails = [row['email'] for row in reader]

    # Massenverifizierung einreichen
    job = client.verify_bulk(emails)
    print(f"Auftrag {job.id} mit {len(emails)} E-Mails eingereicht")

    # Auf Abschluss warten (Polling)
    import time
    while True:
        status = client.get_bulk_job_status(job.id)
        print(f"Fortschritt: {status.processed}/{status.total}")

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

        time.sleep(5)

    # Ergebnisse abrufen und in Ausgabe schreiben
    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"Ergebnisse in {output_file} geschrieben")
    print(f"Gültig: {status.valid}, Ungültig: {status.invalid}")

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

Async-Gleichzeitige Verifizierung

import asyncio
from emailverify import AsyncClient
import os

async def verify_emails_concurrent(emails: list[str], concurrency: int = 10):
    """Mehrere E-Mails mit kontrollierter Parallelität verifizieren."""

    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()

    # Ergebnisse verarbeiten
    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,
    }

# Verwendung
emails = ['user1@example.com', 'user2@example.com', 'user3@example.com']
results = asyncio.run(verify_emails_concurrent(emails))
print(f"Gültig: {len(results['valid'])}, Ungültig: {len(results['invalid'])}")

Typ-Hinweise

Das SDK enthält vollständige Typ-Hinweise für IDE-Unterstützung:

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