Python SDK

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"狀態: {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"任務 ID: {job.id}")
print(f"狀態: {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 Key')
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"已提交任務 {job.id}，包含 {len(emails)} 個郵箱")

    # 等待完成（輪詢）
    import time
    while True:
        status = client.get_bulk_job_status(job.id)
        print(f"進度: {status.processed}/{status.total}")

        if status.status == 'completed':
            break
        elif status.status == 'failed':
            raise Exception(f"任務失敗: {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"結果已寫入 {output_file}")
    print(f"有效: {status.valid}，無效: {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"有效: {len(results['valid'])}，無效: {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

Python SDK

安裝

快速入門

配置

客戶端選項

環境變數

單一郵箱驗證

基本驗證

帶選項的驗證

回應物件

異步使用

上下文管理器

批量驗證

提交批量任務

檢查任務狀態

獲取結果

使用 Webhook

額度管理

錯誤處理

異常類型

重試配置

框架整合

Django

Django 表單驗證

Flask

FastAPI

完整範例

使用者註冊驗證

批量列表清理

異步並發驗證

型別提示

下一步

API 參考文件

整合指南

On this page