FastAPI

고성능 이메일 유효성 검사를 위해 FastAPI 애플리케이션에 EmailVerify를 통합하세요. 이 가이드는 비동기 패턴, 의존성 주입 및 프로덕션 배포를 다룹니다.

설치

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 설정

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

의존성 주입

이메일 검증 서비스

의존성 주입을 사용하여 이메일 검증 서비스를 생성하세요:

from emailverify import EmailVerify
from typing import Optional

class EmailVerificationService:
    """이메일 검증 작업을 위한 서비스."""

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

    async def verify_email(self, email: str) -> dict:
        """단일 이메일 주소 검증."""
        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]:
        """여러 이메일 주소 검증."""
        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:
    """이메일 검증 서비스 의존성 주입."""
    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='캐시 TTL(초)')

응답 모델

일관된 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