Python

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

BillionVerify 公式 Python SDK は、メール検証のための同期および非同期インターフェースを提供します。

インストール

pip install billionverify

クイックスタート

import os
from billionverify import BillionVerify

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

# 単一メールの検証
result = client.verify('user@example.com')
print(f"Status: {result.status}")  # 'valid', 'invalid', 'unknown', または 'catchall'

設定

クライアントオプション

from billionverify import BillionVerify

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

    # オプション設定
    timeout=30,               # リクエストタイムアウト（秒）
    retries=3,                # リトライ回数
    base_url='https://api.billionverify.com/v1',  # カスタム API URL
)

環境変数

# .env
BV_API_KEY=your_api_key_here

import os
from billionverify import BillionVerify

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

単一メール検証

基本的な検証

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

オプション付き検証

result = client.verify(
    'user@example.com',
    smtp_check=True,      # SMTP 検証を有効化
    timeout=5,            # このリクエストのカスタムタイムアウト
)

レスポンスオブジェクト

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

非同期使用

高パフォーマンスアプリケーション向けに非同期クライアントを使用できます：

import asyncio
from billionverify import AsyncBillionVerify

async def main():
    client = AsyncBillionVerify(api_key=os.environ['BV_API_KEY'])

    # 単一検証
    result = await client.verify('user@example.com')
    print(result.status)

    # 複数メールの並行検証
    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())

コンテキストマネージャー

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

一括検証

一括ジョブの送信

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

ジョブステータスの確認

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

結果の取得

results = client.get_bulk_job_results(job.id)

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

Webhook 付き

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

クレジット管理

credits = client.get_credits()

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

エラーハンドリング

例外タイプ

from billionverify import (
    Client,
    BillionVerifyError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
)

try:
    result = client.verify('invalid-email')
except AuthenticationError:
    print('API キーが無効です')
except RateLimitError as e:
    print(f'レート制限に達しました。{e.retry_after} 秒後にリトライしてください')
except ValidationError as e:
    print(f'入力が無効です: {e}')
except BillionVerifyError as e:
    print(f'API エラー: {e}')

リトライ設定

client = BillionVerify(
    api_key=os.environ['BV_API_KEY'],
    retries=5,
    retry_delay=1.0,           # 初期遅延（秒）
    retry_max_delay=30.0,      # リトライ間の最大遅延
)

フレームワーク統合

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': '有効なメールアドレスを入力してください',
            }, status=400)

        if result.result.disposable:
            return JsonResponse({
                'valid': False,
                'message': '使い捨てメールは許可されていません',
            }, status=400)

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

Django フォームバリデーション

# 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('有効なメールアドレスを入力してください')

        if result.result.disposable:
            raise forms.ValidationError('使い捨てメールは許可されていません')

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

完全な例

ユーザー登録バリデーション

from billionverify import BillionVerify
import os

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

def validate_registration_email(email: str) -> dict:
    """ユーザー登録用のメールを検証します。"""
    result = client.verify(email)

    # 無効なメールを拒否
    if result.status == 'invalid':
        return {
            'valid': False,
            'reason': 'invalid_email',
            'message': 'このメールアドレスは無効です',
        }

    # 使い捨てメールを拒否
    if result.result.disposable:
        return {
            'valid': False,
            'reason': 'disposable',
            'message': '恒久的なメールアドレスを使用してください',
        }

    # ロールベースメールについて警告
    if result.result.role:
        return {
            'valid': True,
            'warning': 'role_based',
            'message': '個人メールの使用をお勧めします',
        }

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

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

一括リストクリーニング

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):
    """メールリストをクリーニングして結果をエクスポートします。"""

    # CSV からメールを読み取り
    with open(input_file, 'r') as f:
        reader = csv.DictReader(f)
        emails = [row['email'] for row in reader]

    # 一括検証を送信
    job = client.verify_bulk(emails)
    print(f"Submitted job {job.id} with {len(emails)} emails")

    # 完了を待機（ポーリング）
    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)

    # 結果を取得して出力に書き込み
    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}")

# 使用例
clean_email_list('contacts.csv', 'cleaned_contacts.csv')

非同期並行検証

import asyncio
from billionverify import AsyncBillionVerify
import os

async def verify_emails_concurrent(emails: list[str], concurrency: int = 10):
    """制御された並行性で複数のメールを検証します。"""

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

    # 結果を処理
    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,
    }

# 使用例
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'])}")

型ヒント

SDK には IDE サポートのための完全な型ヒントが含まれています：

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