FastAPI Email Verification

Интегрируйте EmailVerify в ваши FastAPI приложения для высокопроизводительной валидации email. Это руководство охватывает асинхронные паттерны, внедрение зависимостей и развертывание в продакшен.

Установка

Установите FastAPI, EmailVerify и связанные зависимости.

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

Настройка

Переменные окружения

Создайте файл .env для конфигурации:

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

Создайте систему управления конфигурацией с использованием Pydantic Settings:

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

class Settings(BaseSettings):
    """Настройки приложения, загружаемые из переменных окружения."""

    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:
    """Получение кэшированного экземпляра настроек."""
    return Settings()

Внедрение зависимостей

Сервис верификации email

Создайте сервис для верификации email с использованием внедрения зависимостей:

from emailverify import EmailVerify
from typing import Optional

class EmailVerificationService:
    """Сервис для операций верификации email."""

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

    async def verify_email(self, email: str) -> dict:
        """Верификация одного email-адреса."""
        try:
            result = await self.client.verify(email=email)
            return result
        except Exception as e:
            raise ValueError(f"Верификация не удалась: {str(e)}")

    async def verify_bulk(self, emails: list[str]) -> list[dict]:
        """Верификация нескольких email-адресов."""
        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:
    """Внедрение зависимости для сервиса верификации email."""
    return EmailVerificationService(api_key=settings.emailverify_api_key)

Модели Pydantic

Модели запросов

Определите модели запросов для эндпоинтов верификации email:

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

class EmailVerificationRequest(BaseModel):
    """Запрос на верификацию одного email."""
    email: EmailStr = Field(..., description='Email-адрес для верификации')

class BulkVerificationRequest(BaseModel):
    """Запрос на пакетную верификацию email."""
    emails: list[EmailStr] = Field(..., description='Список email-адресов для верификации')
    max_results: Optional[int] = Field(default=None, description='Максимальное количество результатов')

class VerificationQuery(BaseModel):
    """Параметры запроса для верификации email."""
    cache: bool = Field(default=True, description='Использовать кэшированные результаты, если доступны')
    cache_ttl: int = Field(default=86400, description='Время жизни кэша в секундах')

Модели ответов

Определите модели ответов для единообразных API-ответов:

from enum import Enum

class VerificationStatus(str, Enum):
    """Перечисление статусов верификации email."""
    VALID = 'valid'
    INVALID = 'invalid'
    UNKNOWN = 'unknown'
    ACCEPT_ALL = 'accept_all'

class VerificationDetails(BaseModel):
    """Детали верификации email."""
    disposable: bool
    smtp_valid: bool
    format_valid: bool
    email_provider: str
    risk_level: str

class EmailVerificationResponse(BaseModel):
    """Ответ API верификации email."""
    email: str
    status: VerificationStatus
    score: float
    details: VerificationDetails
    cached: bool = False
    timestamp: datetime

class BulkVerificationResponse(BaseModel):
    """Ответ API пакетной верификации."""
    total: int
    verified: int
    failed: int
    results: list[EmailVerificationResponse]
    duration_ms: float

API Эндпоинты

Верификация одного email

Создайте эндпоинт для верификации отдельных email:

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:
    """
    Верификация одного email-адреса.

    Args:
        request: Запрос на верификацию email
        service: Сервис верификации email
        cache_service: Сервис кэширования для хранения результатов

    Returns:
        Ответ с результатом верификации email
    """
    # Сначала проверяем кэш
    cached_result = await cache_service.get(request.email)
    if cached_result:
        return EmailVerificationResponse(
            **cached_result,
            cached=True,
            timestamp=datetime.utcnow()
        )

    try:
        # Верификация email
        result = await service.verify_email(request.email)

        # Кэширование результата
        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 не удалась: {str(e)}'
        )

Пакетная верификация email

Создайте эндпоинт для пакетной верификации email:

@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:
    """
    Пакетная верификация нескольких email-адресов.

    Args:
        request: Запрос на пакетную верификацию
        service: Сервис верификации email
        cache_service: Сервис кэширования

    Returns:
        Ответ с результатами пакетной верификации
    """
    import time

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

    # Ограничение результатов, если указано
    emails = request.emails[:request.max_results] if request.max_results else request.emails

    for email in emails:
        try:
            # Сначала проверяем кэш
            cached = await cache_service.get(email)
            if cached:
                results.append(EmailVerificationResponse(
                    **cached,
                    cached=True,
                    timestamp=datetime.utcnow()
                ))
                continue

            # Верификация email
            result = await service.verify_email(email)

            # Кэширование результата
            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
    )

Валидация форм

Пользовательские валидаторы

Создайте пользовательские валидаторы Pydantic для валидации email:

from pydantic import field_validator, ValidationInfo

class EmailVerificationRequest(BaseModel):
    email: EmailStr

    @field_validator('email')
    @classmethod
    def validate_email_format(cls, v: str) -> str:
        """Валидация формата email."""
        if len(v) > 254:
            raise ValueError('Email-адрес слишком длинный')
        return v.lower()

    @field_validator('email')
    @classmethod
    def validate_email_domain(cls, v: str, info: ValidationInfo) -> str:
        """Проверка, что домен email не одноразовый."""
        domain = v.split('@')[1]

        # Список заблокированных доменов
        blocked_domains = ['tempmail.com', 'guerrillamail.com', '10minutemail.com']

        if domain in blocked_domains:
            raise ValueError('Одноразовые email-адреса не допускаются')

        return v

Middleware

Middleware верификации email

Создайте middleware для автоматической верификации email:

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

class EmailVerificationMiddleware(BaseHTTPMiddleware):
    """Middleware для логирования и отслеживания верификации email."""

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

    async def dispatch(self, request: Request, call_next) -> Response:
        """Обработка запроса и логирование верификаций email."""
        response = await call_next(request)

        if request.url.path == '/api/verify/email':
            # Логирование запроса на верификацию
            print(f'Запрос на верификацию email: {request.url}')

        return response

# Добавление middleware в приложение
app.add_middleware(
    EmailVerificationMiddleware,
    service=get_verification_service()
)

Фоновые задачи

Фоновые задания верификации email

Используйте Celery или Dramatiq для асинхронных фоновых задач:

from fastapi import BackgroundTasks

@router.post('/verify-async')
async def verify_email_async(
    request: EmailVerificationRequest,
    background_tasks: BackgroundTasks,
    service: EmailVerificationService = Depends(get_verification_service)
):
    """Асинхронная верификация email."""

    async def verify_task(email: str):
        """Фоновая задача для верификации email."""
        try:
            result = await service.verify_email(email)
            # Сохранение результата в базу данных
            await store_verification_result(email, result)
        except Exception as e:
            print(f'Фоновая верификация не удалась: {str(e)}')

    background_tasks.add_task(verify_task, request.email)

    return {
        'message': 'Верификация запущена',
        'email': request.email
    }

Кэширование

Интеграция кэширования с Redis

Реализуйте кэширование Redis для результатов верификации:

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

class CacheService:
    """Сервис для кэширования результатов верификации email."""

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

    async def connect(self):
        """Подключение к Redis."""
        self.redis = await redis.from_url(self.redis_url)

    async def disconnect(self):
        """Отключение от Redis."""
        if self.redis:
            await self.redis.close()

    async def get(self, key: str) -> Optional[Any]:
        """Получение значения из кэша."""
        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):
        """Установка значения в кэш с TTL."""
        if not self.redis:
            return

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

    async def delete(self, key: str):
        """Удаление значения из кэша."""
        if not self.redis:
            return

        await self.redis.delete(key)

    async def clear(self):
        """Очистка всего кэша."""
        if not self.redis:
            return

        await self.redis.flushdb()

def get_cache_service(settings: Settings = Depends(get_settings)) -> CacheService:
    """Внедрение зависимости для сервиса кэширования."""
    return CacheService(redis_url=settings.redis_url)

# События запуска и остановки
@app.on_event('startup')
async def startup():
    """Подключение к кэшу при запуске."""
    cache_service = get_cache_service()
    await cache_service.connect()

@app.on_event('shutdown')
async def shutdown():
    """Отключение от кэша при остановке."""
    cache_service = get_cache_service()
    await cache_service.disconnect()

Ограничение частоты запросов

Ограничение частоты с SlowAPI

Реализуйте ограничение частоты запросов с использованием 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)
):
    """Эндпоинт верификации email с ограничением частоты."""
    return await service.verify_email(email_request.email)

Интеграция с базой данных

Модели SQLAlchemy

Определите модели базы данных для хранения результатов верификации:

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):
    """Модель базы данных для результатов верификации email."""

    __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:
    """Фабрика асинхронных сессий базы данных."""

    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):
        """Получение асинхронной сессии базы данных."""
        async with self.SessionLocal() as session:
            yield session

# Инициализация базы данных
db = AsyncSessionLocal(get_settings().database_url)

async def store_verification_result(email: str, result: dict, session):
    """Сохранение результата верификации в базу данных."""
    verification = VerificationResult(
        email=email,
        status=result.get('status'),
        score=result.get('score'),
        **result.get('result', {})
    )
    session.add(verification)
    await session.commit()

Тестирование

Юнит-тесты с Pytest

Тестирование функциональности верификации email:

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

client = TestClient(app)

@pytest.fixture
def mock_service():
    """Мок сервиса верификации email."""
    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):
    """Тест верификации одного email."""
    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():
    """Тест верификации невалидного email."""
    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):
    """Тест пакетной верификации email."""
    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

Развертывание

Конфигурация Docker

Создайте Dockerfile для развертывания:

FROM python:3.11-slim

WORKDIR /app

# Установка зависимостей
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Копирование приложения
COPY . .

# Открытие порта
EXPOSE 8000

# Проверка здоровья
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:8000/health')"

# Запуск приложения
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker Compose

Определите сервисы с 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:

Конфигурация окружения

Настройка продакшен-окружения:

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

Верификация email для Django приложений

Руководство по веб-фреймворкам

Руководство по интеграции для всех веб-фреймворков

Справочник API

Полная документация EmailVerify API

FastAPI Email Verification

Интеграция с Django

Руководство по веб-фреймворкам

Справочник API

On this page