Python SDK
Python email checker SDK. pip install, verify emails with Python. Code examples and async support.
The official EmailVerify SDK for Python provides both synchronous and asynchronous interfaces for email verification.
Installation
pip install emailverifyWith Async Support
pip install emailverify[async]Quick Start
import os
from emailverify import Client
client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])
# Verify a single email
result = client.verify('user@example.com')
print(f"Status: {result.status}") # 'valid', 'invalid', 'unknown', or 'accept_all'Configuration
Client Options
from emailverify import Client
client = Client(
api_key='your-api-key',
# Optional configurations
timeout=30, # Request timeout in seconds
retries=3, # Number of retry attempts
base_url='https://api.emailverify.ai', # Custom API URL
)Environment Variables
# .env
EMAILVERIFY_API_KEY=your_api_key_hereimport os
from emailverify import Client
client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])Single Email Verification
Basic Verification
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,
})Verification with Options
result = client.verify(
'user@example.com',
smtp_check=True, # Enable SMTP verification
timeout=5, # Custom timeout for this request
)Response Object
@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]Async Usage
For high-performance applications, use the async client:
import asyncio
from emailverify import AsyncClient
async def main():
client = AsyncClient(api_key=os.environ['EMAILVERIFY_API_KEY'])
# Single verification
result = await client.verify('user@example.com')
print(result.status)
# Verify multiple emails concurrently
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 AsyncClient(api_key=os.environ['EMAILVERIFY_API_KEY']) as client:
result = await client.verify('user@example.com')Bulk Verification
Submit a Bulk Job
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}")Check 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,
})Get Results
results = client.get_bulk_job_results(job.id)
for result in results:
print(f"{result.email}: {result.status}")With Webhook
job = client.verify_bulk(
emails,
webhook_url='https://your-domain.com/webhooks/emailverify',
)Credit Management
credits = client.get_credits()
print({
'available': credits.available,
'used': credits.used,
'total': credits.total,
})Error Handling
Exception Types
from emailverify import (
Client,
EmailVerifyError,
AuthenticationError,
RateLimitError,
ValidationError,
)
try:
result = client.verify('invalid-email')
except AuthenticationError:
print('Invalid API key')
except RateLimitError as e:
print(f'Rate limited. Retry after {e.retry_after} seconds')
except ValidationError as e:
print(f'Invalid input: {e}')
except EmailVerifyError as e:
print(f'API error: {e}')Retry Configuration
client = Client(
api_key=os.environ['EMAILVERIFY_API_KEY'],
retries=5,
retry_delay=1.0, # Initial delay in seconds
retry_max_delay=30.0, # Max delay between retries
)Framework Integration
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': 'Please enter a valid email address',
}, status=400)
if result.result.disposable:
return JsonResponse({
'valid': False,
'message': 'Disposable emails are not allowed',
}, status=400)
return JsonResponse({'valid': True, 'score': result.score})
except Exception as e:
return JsonResponse({'error': str(e)}, status=500)Django Form Validation
# 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('Please enter a valid email address')
if result.result.disposable:
raise forms.ValidationError('Disposable emails are not allowed')
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()Complete Examples
User Registration Validation
from emailverify import Client
import os
client = Client(api_key=os.environ['EMAILVERIFY_API_KEY'])
def validate_registration_email(email: str) -> dict:
"""Validate email for user registration."""
result = client.verify(email)
# Reject invalid emails
if result.status == 'invalid':
return {
'valid': False,
'reason': 'invalid_email',
'message': 'This email address is invalid',
}
# Reject disposable emails
if result.result.disposable:
return {
'valid': False,
'reason': 'disposable',
'message': 'Please use a permanent email address',
}
# Warn about role-based emails
if result.result.role:
return {
'valid': True,
'warning': 'role_based',
'message': 'Using a personal email is recommended',
}
return {
'valid': True,
'score': result.score,
}
# Usage
validation = validate_registration_email('user@example.com')
if not validation['valid']:
print(validation['message'])Bulk List Cleaning
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):
"""Clean an email list and export results."""
# Read emails from CSV
with open(input_file, 'r') as f:
reader = csv.DictReader(f)
emails = [row['email'] for row in reader]
# Submit bulk verification
job = client.verify_bulk(emails)
print(f"Submitted job {job.id} with {len(emails)} emails")
# Wait for completion (polling)
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)
# Get results and write to output
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}")
# Usage
clean_email_list('contacts.csv', 'cleaned_contacts.csv')Async Concurrent Verification
import asyncio
from emailverify import AsyncClient
import os
async def verify_emails_concurrent(emails: list[str], concurrency: int = 10):
"""Verify multiple emails with controlled concurrency."""
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()
# Process results
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,
}
# Usage
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'])}")Type Hints
The SDK includes full type hints for IDE support:
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