FastAPI

Integrate EmailVerify into your FastAPI applications for high-performance email validation. This guide covers async patterns, dependency injection, and production deployment.

Installation

Install FastAPI, EmailVerify, and related dependencies.

pip install fastapi uvicorn emailverify python-dotenv pydantic redis

poetry add fastapi uvicorn emailverify python-dotenv pydantic redis

uv pip install fastapi uvicorn emailverify python-dotenv pydantic redis

Configuration

Environment Variables

Create a .env file for configuration:

EMAILVERIFY_API_KEY=your_api_key_here
DATABASE_URL=postgresql://user:password@localhost/dbname
REDIS_URL=redis://localhost:6379/0
API_RATE_LIMIT=100
RATE_LIMIT_WINDOW=3600

Pydantic Settings

Create a configuration management system using Pydantic Settings:

from pydantic_settings import BaseSettings
from pydantic import Field
from functools import lru_cache

class Settings(BaseSettings):
    """Application settings loaded from environment variables."""

    emailverify_api_key: str = Field(..., alias='EMAILVERIFY_API_KEY')
    database_url: str = Field(..., alias='DATABASE_URL')
    redis_url: str = Field(default='redis://localhost:6379/0', alias='REDIS_URL')
    api_rate_limit: int = Field(default=100, alias='API_RATE_LIMIT')
    rate_limit_window: int = Field(default=3600, alias='RATE_LIMIT_WINDOW')
    environment: str = Field(default='development')
    debug: bool = Field(default=False)

    class Config:
        env_file = '.env'
        case_sensitive = True

@lru_cache
def get_settings() -> Settings:
    """Get cached settings instance."""
    return Settings()

Dependency Injection

Email Verification Service

Create a service for email verification using dependency injection:

from emailverify import EmailVerify
from typing import Optional

class EmailVerificationService:
    """Service for email verification operations."""

    def __init__(self, api_key: str):
        self.client = EmailVerify(api_key=api_key)

    async def verify_email(self, email: str) -> dict:
        """Verify a single email address."""
        try:
            result = await self.client.verify(email=email)
            return result
        except Exception as e:
            raise ValueError(f"Verification failed: {str(e)}")

    async def verify_bulk(self, emails: list[str]) -> list[dict]:
        """Verify multiple email addresses."""
        results = []
        for email in emails:
            try:
                result = await self.verify_email(email)
                results.append(result)
            except Exception as e:
                results.append({
                    'email': email,
                    'status': 'error',
                    'error': str(e)
                })
        return results

def get_verification_service(settings: Settings = Depends(get_settings)) -> EmailVerificationService:
    """Dependency injection for email verification service."""
    return EmailVerificationService(api_key=settings.emailverify_api_key)

Pydantic Models

Request Models

Define request models for email verification endpoints:

from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class EmailVerificationRequest(BaseModel):
    """Single email verification request."""
    email: EmailStr = Field(..., description='Email address to verify')

class BulkVerificationRequest(BaseModel):
    """Bulk email verification request."""
    emails: list[EmailStr] = Field(..., description='List of email addresses to verify')
    max_results: Optional[int] = Field(default=None, description='Maximum number of results to return')

class VerificationQuery(BaseModel):
    """Query parameters for email verification."""
    cache: bool = Field(default=True, description='Use cached results if available')
    cache_ttl: int = Field(default=86400, description='Cache time-to-live in seconds')

Response Models

Define response models for consistent API responses:

from enum import Enum

class VerificationStatus(str, Enum):
    """Email verification status enumeration."""
    VALID = 'valid'
    INVALID = 'invalid'
    UNKNOWN = 'unknown'
    ACCEPT_ALL = 'accept_all'

class VerificationDetails(BaseModel):
    """Details from email verification."""
    disposable: bool
    smtp_valid: bool
    format_valid: bool
    email_provider: str
    risk_level: str

class EmailVerificationResponse(BaseModel):
    """Email verification API response."""
    email: str
    status: VerificationStatus
    score: float
    details: VerificationDetails
    cached: bool = False
    timestamp: datetime

class BulkVerificationResponse(BaseModel):
    """Bulk verification API response."""
    total: int
    verified: int
    failed: int
    results: list[EmailVerificationResponse]
    duration_ms: float

API Endpoints

Single Email Verification

Create an endpoint for verifying individual emails:

from fastapi import APIRouter, Depends, HTTPException
from datetime import datetime

router = APIRouter(prefix='/api/verify', tags=['email-verification'])

@router.post('/email', response_model=EmailVerificationResponse)
async def verify_email(
    request: EmailVerificationRequest,
    service: EmailVerificationService = Depends(get_verification_service),
    cache_service: CacheService = Depends(get_cache_service)
) -> EmailVerificationResponse:
    """
    Verify a single email address.

    Args:
        request: Email verification request
        service: Email verification service
        cache_service: Cache service for storing results

    Returns:
        Email verification response with status and details
    """
    # Check cache first
    cached_result = await cache_service.get(request.email)
    if cached_result:
        return EmailVerificationResponse(
            **cached_result,
            cached=True,
            timestamp=datetime.utcnow()
        )

    try:
        # Verify email
        result = await service.verify_email(request.email)

        # Cache the result
        await cache_service.set(request.email, result, ttl=86400)

        return EmailVerificationResponse(
            email=request.email,
            status=result.get('status'),
            score=result.get('score', 0),
            details=VerificationDetails(**result.get('result', {})),
            cached=False,
            timestamp=datetime.utcnow()
        )
    except Exception as e:
        raise HTTPException(
            status_code=400,
            detail=f'Email verification failed: {str(e)}'
        )

Bulk Email Verification

Create an endpoint for batch email verification:

@router.post('/bulk', response_model=BulkVerificationResponse)
async def verify_bulk(
    request: BulkVerificationRequest,
    service: EmailVerificationService = Depends(get_verification_service),
    cache_service: CacheService = Depends(get_cache_service)
) -> BulkVerificationResponse:
    """
    Verify multiple email addresses in bulk.

    Args:
        request: Bulk verification request
        service: Email verification service
        cache_service: Cache service

    Returns:
        Bulk verification response with results
    """
    import time

    start_time = time.time()
    results = []
    failed_count = 0

    # Limit results if specified
    emails = request.emails[:request.max_results] if request.max_results else request.emails

    for email in emails:
        try:
            # Try cache first
            cached = await cache_service.get(email)
            if cached:
                results.append(EmailVerificationResponse(
                    **cached,
                    cached=True,
                    timestamp=datetime.utcnow()
                ))
                continue

            # Verify email
            result = await service.verify_email(email)

            # Cache result
            await cache_service.set(email, result)

            results.append(EmailVerificationResponse(
                email=email,
                status=result.get('status'),
                score=result.get('score', 0),
                details=VerificationDetails(**result.get('result', {})),
                cached=False,
                timestamp=datetime.utcnow()
            ))
        except Exception as e:
            failed_count += 1
            results.append(EmailVerificationResponse(
                email=email,
                status='error',
                score=0,
                details=VerificationDetails(...),
                cached=False,
                timestamp=datetime.utcnow()
            ))

    duration_ms = (time.time() - start_time) * 1000

    return BulkVerificationResponse(
        total=len(emails),
        verified=len(results) - failed_count,
        failed=failed_count,
        results=results,
        duration_ms=duration_ms
    )

Form Validation

Custom Validators

Create custom Pydantic validators for email validation:

from pydantic import field_validator, ValidationInfo

class EmailVerificationRequest(BaseModel):
    email: EmailStr

    @field_validator('email')
    @classmethod
    def validate_email_format(cls, v: str) -> str:
        """Validate email format."""
        if len(v) > 254:
            raise ValueError('Email address too long')
        return v.lower()

    @field_validator('email')
    @classmethod
    def validate_email_domain(cls, v: str, info: ValidationInfo) -> str:
        """Validate email domain is not disposable."""
        domain = v.split('@')[1]

        # List of blocked domains
        blocked_domains = ['tempmail.com', 'guerrillamail.com', '10minutemail.com']

        if domain in blocked_domains:
            raise ValueError('Disposable email addresses are not allowed')

        return v

Middleware

Email Verification Middleware

Create middleware for automatic email verification:

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

class EmailVerificationMiddleware(BaseHTTPMiddleware):
    """Middleware for email verification logging and tracking."""

    def __init__(self, app, service: EmailVerificationService):
        super().__init__(app)
        self.service = service

    async def dispatch(self, request: Request, call_next) -> Response:
        """Process request and log email verifications."""
        response = await call_next(request)

        if request.url.path == '/api/verify/email':
            # Log verification request
            print(f'Email verification request: {request.url}')

        return response

# Add middleware to app
app.add_middleware(
    EmailVerificationMiddleware,
    service=get_verification_service()
)

Background Tasks

Email Verification Background Jobs

Use Celery or Dramatiq for async background tasks:

from fastapi import BackgroundTasks

@router.post('/verify-async')
async def verify_email_async(
    request: EmailVerificationRequest,
    background_tasks: BackgroundTasks,
    service: EmailVerificationService = Depends(get_verification_service)
):
    """Verify email asynchronously."""

    async def verify_task(email: str):
        """Background task for email verification."""
        try:
            result = await service.verify_email(email)
            # Store result in database
            await store_verification_result(email, result)
        except Exception as e:
            print(f'Background verification failed: {str(e)}')

    background_tasks.add_task(verify_task, request.email)

    return {
        'message': 'Verification started',
        'email': request.email
    }

Caching

Redis Caching Integration

Implement Redis caching for verification results:

import redis.asyncio as redis
from typing import Optional, Any

class CacheService:
    """Service for caching email verification results."""

    def __init__(self, redis_url: str):
        self.redis_url = redis_url
        self.redis: Optional[redis.Redis] = None

    async def connect(self):
        """Connect to Redis."""
        self.redis = await redis.from_url(self.redis_url)

    async def disconnect(self):
        """Disconnect from Redis."""
        if self.redis:
            await self.redis.close()

    async def get(self, key: str) -> Optional[Any]:
        """Get value from cache."""
        if not self.redis:
            return None

        value = await self.redis.get(key)
        return json.loads(value) if value else None

    async def set(self, key: str, value: Any, ttl: int = 86400):
        """Set value in cache with TTL."""
        if not self.redis:
            return

        await self.redis.setex(
            key,
            ttl,
            json.dumps(value)
        )

    async def delete(self, key: str):
        """Delete value from cache."""
        if not self.redis:
            return

        await self.redis.delete(key)

    async def clear(self):
        """Clear all cache."""
        if not self.redis:
            return

        await self.redis.flushdb()

def get_cache_service(settings: Settings = Depends(get_settings)) -> CacheService:
    """Dependency injection for cache service."""
    return CacheService(redis_url=settings.redis_url)

# Startup and shutdown events
@app.on_event('startup')
async def startup():
    """Connect to cache on startup."""
    cache_service = get_cache_service()
    await cache_service.connect()

@app.on_event('shutdown')
async def shutdown():
    """Disconnect from cache on shutdown."""
    cache_service = get_cache_service()
    await cache_service.disconnect()

Rate Limiting

SlowAPI Rate Limiting

Implement rate limiting using SlowAPI:

pip install slowapi

from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@router.post('/email')
@limiter.limit('100/minute')
async def verify_email(
    request: Request,
    email_request: EmailVerificationRequest,
    service: EmailVerificationService = Depends(get_verification_service)
):
    """Rate-limited email verification endpoint."""
    return await service.verify_email(email_request.email)

Database Integration

SQLAlchemy Models

Define database models for storing verification results:

from sqlalchemy import Column, String, DateTime, Float, Boolean
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class VerificationResult(Base):
    """Database model for email verification results."""

    __tablename__ = 'verification_results'

    email = Column(String(254), primary_key=True)
    status = Column(String(50))
    score = Column(Float)
    disposable = Column(Boolean)
    smtp_valid = Column(Boolean)
    format_valid = Column(Boolean)
    email_provider = Column(String(100))
    risk_level = Column(String(20))
    verified_at = Column(DateTime, default=datetime.utcnow)
    cached_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

class AsyncSessionLocal:
    """Async database session factory."""

    def __init__(self, database_url: str):
        from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
        from sqlalchemy.orm import sessionmaker

        self.engine = create_async_engine(database_url, echo=False)
        self.SessionLocal = sessionmaker(
            self.engine,
            class_=AsyncSession,
            expire_on_commit=False
        )

    async def get_session(self):
        """Get async database session."""
        async with self.SessionLocal() as session:
            yield session

# Initialize database
db = AsyncSessionLocal(get_settings().database_url)

async def store_verification_result(email: str, result: dict, session):
    """Store verification result in database."""
    verification = VerificationResult(
        email=email,
        status=result.get('status'),
        score=result.get('score'),
        **result.get('result', {})
    )
    session.add(verification)
    await session.commit()

Testing

Unit Tests with Pytest

Test email verification functionality:

import pytest
from fastapi.testclient import TestClient
from unittest.mock import AsyncMock, patch

client = TestClient(app)

@pytest.fixture
def mock_service():
    """Mock email verification service."""
    service = AsyncMock()
    service.verify_email = AsyncMock(return_value={
        'status': 'valid',
        'score': 1.0,
        'result': {
            'disposable': False,
            'smtp_valid': True,
            'format_valid': True,
            'email_provider': 'gmail',
            'risk_level': 'low'
        }
    })
    return service

def test_verify_email(mock_service):
    """Test single email verification."""
    with patch('services.get_verification_service', return_value=mock_service):
        response = client.post(
            '/api/verify/email',
            json={'email': 'test@example.com'}
        )

    assert response.status_code == 200
    assert response.json()['status'] == 'valid'

def test_verify_email_invalid():
    """Test invalid email verification."""
    response = client.post(
        '/api/verify/email',
        json={'email': 'invalid-email'}
    )

    assert response.status_code == 422

@pytest.mark.asyncio
async def test_bulk_verification(mock_service):
    """Test bulk email verification."""
    with patch('services.get_verification_service', return_value=mock_service):
        response = client.post(
            '/api/verify/bulk',
            json={'emails': ['test1@example.com', 'test2@example.com']}
        )

    assert response.status_code == 200
    assert response.json()['total'] == 2

Deployment

Docker Configuration

Create a Dockerfile for deployment:

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:8000/health')"

# Run application
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker Compose

Define services with Docker Compose:

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql://user:password@db:5432/emailverify
      REDIS_URL: redis://redis:6379/0
      EMAILVERIFY_API_KEY: ${EMAILVERIFY_API_KEY}
    depends_on:
      - db
      - redis

  db:
    image: postgres:15
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: emailverify
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  redis_data:

Environment Configuration

Production environment setup:

# .env.production
EMAILVERIFY_API_KEY=sk_live_xxxxxxxxxxxxx
DATABASE_URL=postgresql://user:password@prod-db.example.com/emailverify
REDIS_URL=redis://prod-redis.example.com:6379/0
ENVIRONMENT=production
DEBUG=false
API_RATE_LIMIT=1000
RATE_LIMIT_WINDOW=3600

Django Integration

Email verification for Django applications

Web Frameworks Guide

Integration guide for all web frameworks

API Reference

Complete EmailVerify API documentation

FastAPI

Django Integration

Web Frameworks Guide

API Reference

On this page