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

Retry 配置

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

框架集成

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

额度管理

错误处理

异常类型

Retry 配置

框架集成

Django

Django 表单验证

Flask

FastAPI

完整示例

用户注册验证

批量列表清理

异步并发验证

类型提示

下一步

API 参考

集成指南

On this page