Python SDK

BillionVerify SDK für Python. Installation per pip, synchrone und asynchrone E-Mail-Verifizierung mit Codebeispielen.

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

Installation

pip install billionverify

Schnellstart

import os
from billionverify import BillionVerify

client = BillionVerify(api_key=os.environ['BV_API_KEY'])

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

Konfiguration

Client-Optionen

from billionverify import BillionVerify

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

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

Umgebungsvariablen

# .env
BV_API_KEY=your_api_key_here

import os
from billionverify import BillionVerify

client = BillionVerify(api_key=os.environ['BV_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', 'catchall'
    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 billionverify import AsyncBillionVerify

async def main():
    client = AsyncBillionVerify(api_key=os.environ['BV_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 AsyncBillionVerify(api_key=os.environ['BV_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/billionverify',
)

Guthaben-Verwaltung

credits = client.get_credits()

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

Fehlerbehandlung

Ausnahmetypen

from billionverify import (
    Client,
    BillionVerifyError,
    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 BillionVerifyError as e:
    print(f'API-Fehler: {e}')

Wiederholungskonfiguration

client = BillionVerify(
    api_key=os.environ['BV_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 billionverify import BillionVerify
import os

client = BillionVerify(api_key=os.environ['BV_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 billionverify import BillionVerify
import os

client = BillionVerify(api_key=os.environ['BV_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 billionverify import BillionVerify
import os

app = Flask(__name__)
client = BillionVerify(api_key=os.environ['BV_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 billionverify import AsyncBillionVerify
import os

app = FastAPI()
client = AsyncBillionVerify(api_key=os.environ['BV_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 billionverify import BillionVerify
import os

client = BillionVerify(api_key=os.environ['BV_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 billionverify import BillionVerify
import os

client = BillionVerify(api_key=os.environ['BV_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 billionverify import AsyncBillionVerify
import os

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

    client = AsyncBillionVerify(api_key=os.environ['BV_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 billionverify import BillionVerify, VerificationResult, BulkJob
from typing import Optional

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