EmailVerify LogoEmailVerify
Guías de IntegraciónFrameworks Web

FastAPI

Email checker for FastAPI. Python email verification in FastAPI endpoints.

Integre EmailVerify en sus aplicaciones FastAPI para validación de correo de alto rendimiento. Esta guía cubre patrones asíncronos, inyección de dependencias y despliegue en producción.

Instalación

Instale FastAPI, EmailVerify y las dependencias relacionadas.

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

Configuración

Variables de Entorno

Cree un archivo .env para la configuración:

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

Configuración con Pydantic Settings

Cree un sistema de gestión de configuración usando 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_sinsitive = True

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

Inyección de Dependencias

Servicio de Verificación de Correo

Cree un servicio para verificación de correo usando inyección de dependencias:

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)

Modelos Pydantic

Modelos de Solicitud

Defina modelos de solicitud para los endpoints de verificación de correo:

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

Modelos de Respuesta

Defina modelos de respuesta para respuestas API consistentes:

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

Endpoints de API

Verificación de Correo Individual

Cree un endpoint para verificar correos individuales:

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

Verificación de Correo en Lote

Cree un endpoint para verificación de correo por lotes:

@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
    )

Validación de Formularios

Validadores Personalizados

Cree validadores Pydantic personalizados para validación de correo:

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

Middleware de Verificación de Correo

Cree middleware para verificación de correo automática:

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()
)

Tareas en Segundo Plano

Trabajos de Verificación de Correo en Segundo Plano

Use Celery o Dramatiq para tareas asíncronas en segundo plano:

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
    }

Almacenamiento en Caché

Integración de Almacenamiento en Caché con Redis

Implemente almacenamiento en caché con Redis para los resultados de verificación:

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()

Limitación de Tasa

Limitación de Tasa con SlowAPI

Implemente limitación de tasa usando 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)

Integración con Base de Datos

Modelos SQLAlchemy

Defina modelos de base de datos para almacenar resultados de verificación:

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()

Pruebas

Pruebas Unitarias con Pytest

Pruebe la funcionalidad de verificación de correo:

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

Despliegue

Configuración de Docker

Cree un Dockerfile para el despliegue:

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

Defina servicios con 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:

Configuración del Entorno

Configuración del entorno de producción:

# .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

Recursos Relacionados

Integración con Django

Verificación de correo para aplicaciones Django

Guía de Frameworks Web

Guía de integración para todos los frameworks web

Referencia API

Documentación completa de la API de EmailVerify

Django

Email checker for Django. Python email verification in Django views and forms.

Gin

Email checker for Go Gin. Email verification in Gin HTTP handlers.

On this page

InstalaciónConfiguraciónVariables de EntornoConfiguración con Pydantic SettingsInyección de DependenciasServicio de Verificación de CorreoModelos PydanticModelos de SolicitudModelos de RespuestaEndpoints de APIVerificación de Correo IndividualVerificación de Correo en LoteValidación de FormulariosValidadores PersonalizadosMiddlewareMiddleware de Verificación de CorreoTareas en Segundo PlanoTrabajos de Verificación de Correo en Segundo PlanoAlmacenamiento en CachéIntegración de Almacenamiento en Caché con RedisLimitación de TasaLimitación de Tasa con SlowAPIIntegración con Base de DatosModelos SQLAlchemyPruebasPruebas Unitarias con PytestDespliegueConfiguración de DockerDocker ComposeConfiguración del EntornoRecursos Relacionados