Python SDK
Python email checker SDK. pip install, verify emails with Python. Code examples and async support.
官方 EmailVerify Python SDK 提供同步和異步介面進行郵箱驗證。
安裝
pip install emailverify支援異步
pip install emailverify[async]快速入門
import os
from emailverify import Client
client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])
# 驗證單個郵箱
result = client.verify('user@example.com')
print(f"狀態: {result.status}") # 'valid'、'invalid'、'unknown' 或 'accept_all'配置
客戶端選項
from emailverify import Client
client = Client(
api_key='your-api-key',
# 可選配置
timeout=30, # 請求逾時（秒）
retries=3, # 重試次數
base_url='https://api.emailverify.ai', # 自訂 API URL
)環境變數
# .env
EMAILVERIFY_API_KEY=your_api_key_hereimport os
from emailverify import Client
client = Client(api_key=os.environ['EMAILVERIFY_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'、'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]異步使用
對於高效能應用程式，使用異步客戶端：
import asyncio
from emailverify import AsyncClient
async def main():
client = AsyncClient(api_key=os.environ['EMAILVERIFY_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 AsyncClient(api_key=os.environ['EMAILVERIFY_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/emailverify',
)額度管理
credits = client.get_credits()
print({
'available': credits.available,
'used': credits.used,
'total': credits.total,
})錯誤處理
異常類型
from emailverify import (
Client,
EmailVerifyError,
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 EmailVerifyError as e:
print(f'API 錯誤: {e}')重試配置
client = Client(
api_key=os.environ['EMAILVERIFY_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 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': '請輸入有效的郵箱地址',
}, 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 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('請輸入有效的郵箱地址')
if result.result.disposable:
raise forms.ValidationError('不允許使用一次性郵箱')
return emailFlask
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)}), 500FastAPI
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()完整範例
使用者註冊驗證
from emailverify import Client
import os
client = Client(api_key=os.environ['EMAILVERIFY_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 emailverify import Client
import os
client = Client(api_key=os.environ['EMAILVERIFY_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 emailverify import AsyncClient
import os
async def verify_emails_concurrent(emails: list[str], concurrency: int = 10):
"""以受控的並發方式驗證多個郵箱。"""
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()
# 處理結果
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 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