Python Email Validation
Complete guide to email validation in Python
What You'll Learn
- Install and configure VerifyForge in Python
- Validate emails in Python scripts
- Build validation functions
- Handle errors gracefully
- Process bulk validations
- Best practices for production
Prerequisites
- Python 3.8+ installed
- Basic knowledge of Python
- VerifyForge API key (get free key)
Installation
pip install verifyforge
Quick Start
from verifyforge import VerifyForge
client = VerifyForge(api_key="your_api_key")
result = client.validate("user@example.com")
if result.data.is_valid:
print(f"✓ Email is valid!")
print(f"Reachability: {result.data.reachability}")
Tutorial 1: Basic Validation Script
# validate_email.py
import sys
from verifyforge import VerifyForge, VerifyForgeError
def validate_email(email: str) -> dict:
"""Validate a single email address"""
client = VerifyForge(api_key="your_api_key")
try:
result = client.validate(email)
return {
"email": result.data.email,
"is_valid": result.data.is_valid,
"reachability": result.data.reachability,
"disposable": result.data.disposable,
"role_account": result.data.role_account,
"free_provider": result.data.free_provider,
"suggestion": result.data.suggestion,
}
except VerifyForgeError as e:
print(f"Validation error: {e}")
return None
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python validate_email.py <email>")
sys.exit(1)
email = sys.argv[1]
result = validate_email(email)
if result:
print(f"\nValidation Results:")
print(f"Email: {result['email']}")
print(f"Valid: {'✓' if result['is_valid'] else '✗'}")
print(f"Reachability: {result['reachability']}")
print(f"Disposable: {result['disposable']}")
if result['suggestion']:
print(f"Suggestion: {result['suggestion']}")
Run:
python validate_email.py user@example.com
Tutorial 2: Validation with Caching
# email_validator.py
import os
from functools import lru_cache
from verifyforge import VerifyForge
client = VerifyForge(api_key=os.environ.get("VERIFYFORGE_API_KEY"))
@lru_cache(maxsize=1000)
def validate_email_cached(email: str) -> dict:
"""Validate email with caching"""
result = client.validate(email)
return {
"is_valid": result.data.is_valid,
"reachability": result.data.reachability,
"disposable": result.data.disposable,
}
def is_valid_email(email: str) -> bool:
"""Simple boolean check"""
try:
result = validate_email_cached(email)
return result["is_valid"]
except Exception:
return False
Tutorial 3: Bulk Validation for CSV
# bulk_validate_csv.py
import csv
from verifyforge import VerifyForge
def validate_csv(input_file: str, output_file: str, batch_size: int = 100):
"""Validate emails from CSV file"""
client = VerifyForge(api_key="your_api_key")
# Read emails from CSV
emails = []
with open(input_file, 'r') as f:
reader = csv.DictReader(f)
for row in reader:
emails.append(row['email'])
print(f"Found {len(emails)} emails to validate")
# Validate in batches
all_results = []
for i in range(0, len(emails), batch_size):
batch = emails[i:i + batch_size]
print(f"Validating batch {i//batch_size + 1}...")
try:
result = client.validate_bulk(batch)
for validation in result.results:
all_results.append({
'email': validation.email,
'is_valid': validation.is_valid,
'reachable': validation.reachable,
})
except Exception as e:
print(f"Batch failed: {e}")
# Write results to CSV
with open(output_file, 'w', newline='') as f:
fieldnames = ['email', 'is_valid', 'reachable']
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(all_results)
print(f"Results written to {output_file}")
if __name__ == "__main__":
validate_csv('emails.csv', 'validated_emails.csv')
Tutorial 4: API with FastAPI
# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from verifyforge import (
VerifyForge,
AuthenticationError,
InsufficientCreditsError,
ValidationError
)
import os
app = FastAPI()
client = VerifyForge(api_key=os.environ.get("VERIFYFORGE_API_KEY"))
class EmailRequest(BaseModel):
email: EmailStr
class BulkEmailRequest(BaseModel):
emails: list[EmailStr]
@app.post("/api/validate")
async def validate_email(request: EmailRequest):
try:
result = client.validate(request.email)
return {
"success": True,
"data": {
"email": result.data.email,
"is_valid": result.data.is_valid,
"reachability": result.data.reachability,
"disposable": result.data.disposable,
"role_account": result.data.role_account,
"suggestion": result.data.suggestion,
}
}
except AuthenticationError:
raise HTTPException(status_code=401, detail="Invalid API key")
except InsufficientCreditsError:
raise HTTPException(status_code=402, detail="Insufficient credits")
except ValidationError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
raise HTTPException(status_code=500, detail="Validation failed")
@app.post("/api/validate-bulk")
async def validate_bulk(request: BulkEmailRequest):
if len(request.emails) > 100:
raise HTTPException(
status_code=400,
detail="Maximum 100 emails per request"
)
try:
result = client.validate_bulk(request.emails)
return {
"success": True,
"data": {
"results": [
{
"email": r.email,
"is_valid": r.is_valid,
"reachable": r.reachable
}
for r in result.results
],
"summary": {
"total": result.summary.total,
"valid": result.summary.valid,
"invalid": result.summary.invalid,
}
}
}
except Exception as e:
raise HTTPException(status_code=500, detail="Bulk validation failed")
Run:
uvicorn main:app --reload
Tutorial 5: Pandas DataFrame Validation
# validate_dataframe.py
import pandas as pd
from verifyforge import VerifyForge
def validate_dataframe(df: pd.DataFrame, email_column: str = 'email') -> pd.DataFrame:
"""Add validation columns to DataFrame"""
client = VerifyForge(api_key="your_api_key")
emails = df[email_column].tolist()
# Validate in batches
batch_size = 100
validations = []
for i in range(0, len(emails), batch_size):
batch = emails[i:i + batch_size]
result = client.validate_bulk(batch)
validations.extend(result.results)
# Add validation columns
df['is_valid'] = [v.is_valid for v in validations]
df['reachability'] = [v.reachable for v in validations]
return df
# Usage
if __name__ == "__main__":
# Read CSV
df = pd.read_csv('subscribers.csv')
# Validate
df = validate_dataframe(df)
# Filter valid emails
valid_df = df[df['is_valid'] == True]
# Save
valid_df.to_csv('valid_subscribers.csv', index=False)
print(f"Valid emails: {len(valid_df)}/{len(df)}")
Best Practices
1. Environment Variables
import os
from dotenv import load_dotenv
load_dotenv()
client = VerifyForge(api_key=os.environ.get("VERIFYFORGE_API_KEY"))
# .env
VERIFYFORGE_API_KEY=your_api_key_here
2. Context Manager
from verifyforge import VerifyForge
with VerifyForge(api_key="your_api_key") as client:
result = client.validate("test@example.com")
print(result.data.is_valid)
3. Error Handling
from verifyforge import (
VerifyForge,
AuthenticationError,
InsufficientCreditsError,
ValidationError
)
client = VerifyForge(api_key="your_api_key")
try:
result = client.validate("test@example.com")
except AuthenticationError:
print("Invalid API key")
except InsufficientCreditsError:
print("Out of credits")
except ValidationError as e:
print(f"Validation error: {e.message}")
except Exception as e:
print(f"Unexpected error: {e}")
4. Async Validation
import asyncio
from concurrent.futures import ThreadPoolExecutor
from verifyforge import VerifyForge
client = VerifyForge(api_key="your_api_key")
async def validate_async(emails: list[str]) -> list:
"""Validate emails asynchronously"""
loop = asyncio.get_event_loop()
with ThreadPoolExecutor() as executor:
futures = [
loop.run_in_executor(executor, client.validate, email)
for email in emails
]
results = await asyncio.gather(*futures)
return results
# Usage
emails = ["test1@example.com", "test2@example.com"]
results = asyncio.run(validate_async(emails))
5. Retry Logic
import time
from verifyforge import VerifyForge, VerifyForgeError
def validate_with_retry(email: str, max_retries: int = 3) -> dict:
"""Validate with exponential backoff"""
client = VerifyForge(api_key="your_api_key")
for attempt in range(max_retries):
try:
result = client.validate(email)
return result.data
except VerifyForgeError as e:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt # Exponential backoff
print(f"Retry {attempt + 1} in {wait_time}s...")
time.sleep(wait_time)
Complete Example: Email List Cleaner
# email_cleaner.py
import csv
import sys
from typing import List, Dict
from verifyforge import VerifyForge
class EmailCleaner:
def __init__(self, api_key: str):
self.client = VerifyForge(api_key=api_key)
def clean_email_list(
self,
emails: List[str],
allow_disposable: bool = False,
allow_risky: bool = False
) -> Dict[str, List[str]]:
"""Clean email list and categorize results"""
valid_emails = []
invalid_emails = []
risky_emails = []
disposable_emails = []
# Validate in batches
batch_size = 100
for i in range(0, len(emails), batch_size):
batch = emails[i:i + batch_size]
try:
result = self.client.validate_bulk(batch)
for validation in result.results:
if not validation.is_valid:
invalid_emails.append(validation.email)
elif validation.is_valid and validation.disposable:
if allow_disposable:
valid_emails.append(validation.email)
else:
disposable_emails.append(validation.email)
elif validation.reachable == 'risky':
if allow_risky:
valid_emails.append(validation.email)
else:
risky_emails.append(validation.email)
else:
valid_emails.append(validation.email)
except Exception as e:
print(f"Batch {i//batch_size + 1} failed: {e}")
return {
"valid": valid_emails,
"invalid": invalid_emails,
"risky": risky_emails,
"disposable": disposable_emails,
}
def clean_csv(self, input_file: str, output_prefix: str = "cleaned"):
"""Clean CSV file and output categorized results"""
# Read emails
with open(input_file, 'r') as f:
reader = csv.DictReader(f)
emails = [row['email'] for row in reader]
print(f"Cleaning {len(emails)} emails...")
# Clean
results = self.clean_email_list(emails)
# Write results
for category, email_list in results.items():
if email_list:
filename = f"{output_prefix}_{category}.csv"
with open(filename, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['email'])
for email in email_list:
writer.writerow([email])
print(f"Wrote {len(email_list)} {category} emails to {filename}")
if __name__ == "__main__":
cleaner = EmailCleaner(api_key="your_api_key")
cleaner.clean_csv("email_list.csv")
Troubleshooting
Problem: Module not found
Solution: Install SDK: pip install verifyforge
Problem: API key error
Solution: Set environment variable: export VERIFYFORGE_API_KEY=your_key
Problem: Timeout errors
Solution: Increase timeout:
client = VerifyForge(api_key="your_key", timeout=60)