FastAPI

将 EmailVerify 整合到您的 FastAPI 應用中以实現高性能電子郵件驗證。本指南涵盖异步模式、依賴注入和生產部署。

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 模型

請求模型

為電子郵件驗證端点定義請求模型：

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

class EmailVerificationRequest(BaseModel):
    """單個電子郵件驗證請求。"""
    email: EmailStr = Field(..., description='要驗證的電子郵件地址')

class BulkVerificationRequest(BaseModel):
    """批量電子郵件驗證請求。"""
    emails: list[EmailStr] = Field(..., description='要驗證的電子郵件地址列表')
    max_results: Optional[int] = Field(default=None, description='傳回的最大結果数')

class VerificationQuery(BaseModel):
    """電子郵件驗證的查询参数。"""
    cache: bool = Field(default=True, description='如果可用，使用快取結果')
    cache_ttl: int = Field(default=86400, description='快取的生存時间（秒）')

回應模型

為一致的 API 回應定義回應模型：

from enum import Enum

class VerificationStatus(str, Enum):
    """電子郵件驗證状态枚舉。"""
    VALID = 'valid'
    INVALID = 'invalid'
    UNKNOWN = 'unknown'
    ACCEPT_ALL = 'accept_all'

class VerificationDetails(BaseModel):
    """来自電子郵件驗證的詳細信息。"""
    disposable: bool
    smtp_valid: bool
    format_valid: bool
    email_provider: str
    risk_level: str

class EmailVerificationResponse(BaseModel):
    """電子郵件驗證 API 回應。"""
    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 端点

單個電子郵件驗證

為驗證單個電子郵件创建端点：

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:
    """
    驗證單個電子郵件地址。

    Args:
        request: 電子郵件驗證請求
        service: 電子郵件驗證服务
        cache_service: 用於存儲結果的快取服务

    Returns:
        包含状态和詳細信息的電子郵件驗證回應
    """
    # 首先檢查快取
    cached_result = await cache_service.get(request.email)
    if cached_result:
        return EmailVerificationResponse(
            **cached_result,
            cached=True,
            timestamp=datetime.utcnow()
        )

    try:
        # 驗證電子郵件
        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'電子郵件驗證失敗：{str(e)}'
        )

批量電子郵件驗證

為批量電子郵件驗證创建端点：

@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:
    """
    批量驗證多個電子郵件地址。

    Args:
        request: 批量驗證請求
        service: 電子郵件驗證服务
        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

            # 驗證電子郵件
            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 驗證器：

from pydantic import field_validator, ValidationInfo

class EmailVerificationRequest(BaseModel):
    email: EmailStr

    @field_validator('email')
    @classmethod
    def validate_email_format(cls, v: str) -> str:
        """驗證電子郵件格式。"""
        if len(v) > 254:
            raise ValueError('電子郵件地址过长')
        return v.lower()

    @field_validator('email')
    @classmethod
    def validate_email_domain(cls, v: str, info: ValidationInfo) -> str:
        """驗證電子郵件域名不是一次性的。"""
        domain = v.split('@')[1]

        # 被阻止的域名列表
        blocked_domains = ['tempmail.com', 'guerrillamail.com', '10minutemail.com']

        if domain in blocked_domains:
            raise ValueError('不允许使用一次性電子郵件地址')

        return v

中间件

電子郵件驗證中间件

為自动電子郵件驗證创建中间件：

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

class EmailVerificationMiddleware(BaseHTTPMiddleware):
    """用於電子郵件驗證日志記錄和跟踪的中间件。"""

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

    async def dispatch(self, request: Request, call_next) -> Response:
        """處理請求并記錄電子郵件驗證。"""
        response = await call_next(request)

        if request.url.path == '/api/verify/email':
            # 記錄驗證請求
            print(f'電子郵件驗證請求：{request.url}')

        return response

# 向應用添加中间件
app.add_middleware(
    EmailVerificationMiddleware,
    service=get_verification_service()
)

后台任务

電子郵件驗證后台作業

使用 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)
):
    """异步驗證電子郵件。"""

    async def verify_task(email: str):
        """用於電子郵件驗證的后台任务。"""
        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:
    """用於快取電子郵件驗證結果的服务。"""

    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)
):
    """速率限制的電子郵件驗證端点。"""
    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):
    """用於電子郵件驗證結果的資料函式庫模型。"""

    __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 進行單元测試

测試電子郵件驗證功能：

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

client = TestClient(app)

@pytest.fixture
def mock_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):
    """测試單個電子郵件驗證。"""
    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():
    """测試無效電子郵件驗證。"""
    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):
    """测試批量電子郵件驗證。"""
    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