FastAPI

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

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

class Settings(BaseSettings):
    """Applicatie-instellingen geladen uit omgevingsvariabelen."""

    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:
    """Haal gecachte settings instantie op."""
    return Settings()

from emailverify import EmailVerify
from typing import Optional

class EmailVerificationService:
    """Service voor e-mailverificatie operaties."""

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

    async def verify_email(self, email: str) -> dict:
        """Verifieer een enkel e-mailadres."""
        try:
            result = await self.client.verify(email=email)
            return result
        except Exception as e:
            raise ValueError(f"Verificatie mislukt: {str(e)}")

    async def verify_bulk(self, emails: list[str]) -> list[dict]:
        """Verifieer meerdere e-mailadressen."""
        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 voor e-mailverificatie service."""
    return EmailVerificationService(api_key=settings.emailverify_api_key)

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

class EmailVerificationRequest(BaseModel):
    """Enkel e-mailverificatie verzoek."""
    email: EmailStr = Field(..., description='E-mailadres om te verifieren')

class BulkVerificationRequest(BaseModel):
    """Bulk e-mailverificatie verzoek."""
    emails: list[EmailStr] = Field(..., description='Lijst van e-mailadressen om te verifieren')
    max_results: Optional[int] = Field(default=None, description='Maximum aantal resultaten om te retourneren')

class VerificationQuery(BaseModel):
    """Query parameters voor e-mailverificatie."""
    cache: bool = Field(default=True, description='Gebruik gecachte resultaten indien beschikbaar')
    cache_ttl: int = Field(default=86400, description='Cache time-to-live in seconden')

from enum import Enum

class VerificationStatus(str, Enum):
    """E-mailverificatie status enumeratie."""
    VALID = 'valid'
    INVALID = 'invalid'
    UNKNOWN = 'unknown'
    ACCEPT_ALL = 'accept_all'

class VerificationDetails(BaseModel):
    """Details van e-mailverificatie."""
    disposable: bool
    smtp_valid: bool
    format_valid: bool
    email_provider: str
    risk_level: str

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

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

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:
    """
    Verifieer een enkel e-mailadres.

    Args:
        request: E-mailverificatie verzoek
        service: E-mailverificatie service
        cache_service: Cache service voor opslaan van resultaten

    Returns:
        E-mailverificatie response met status en details
    """
    # Controleer cache eerst
    cached_result = await cache_service.get(request.email)
    if cached_result:
        return EmailVerificationResponse(
            **cached_result,
            cached=True,
            timestamp=datetime.utcnow()
        )

    try:
        # Verifieer e-mail
        result = await service.verify_email(request.email)

        # Cache het resultaat
        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'E-mailverificatie mislukt: {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:
    """
    Verifieer meerdere e-mailadressen in bulk.

    Args:
        request: Bulk verificatie verzoek
        service: E-mailverificatie service
        cache_service: Cache service

    Returns:
        Bulk verificatie response met resultaten
    """
    import time

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

    # Beperk resultaten indien gespecificeerd
    emails = request.emails[:request.max_results] if request.max_results else request.emails

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

            # Verifieer e-mail
            result = await service.verify_email(email)

            # Cache resultaat
            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
    )

from pydantic import field_validator, ValidationInfo

class EmailVerificationRequest(BaseModel):
    email: EmailStr

    @field_validator('email')
    @classmethod
    def validate_email_format(cls, v: str) -> str:
        """Valideer e-mailformaat."""
        if len(v) > 254:
            raise ValueError('E-mailadres te lang')
        return v.lower()

    @field_validator('email')
    @classmethod
    def validate_email_domain(cls, v: str, info: ValidationInfo) -> str:
        """Valideer dat e-maildomein niet wegwerp is."""
        domain = v.split('@')[1]

        # Lijst van geblokkeerde domeinen
        blocked_domains = ['tempmail.com', 'guerrillamail.com', '10minutemail.com']

        if domain in blocked_domains:
            raise ValueError('Wegwerp e-mailadressen zijn niet toegestaan')

        return v

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

class EmailVerificationMiddleware(BaseHTTPMiddleware):
    """Middleware voor e-mailverificatie logging en tracking."""

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

    async def dispatch(self, request: Request, call_next) -> Response:
        """Verwerk verzoek en log e-mailverificaties."""
        response = await call_next(request)

        if request.url.path == '/api/verify/email':
            # Log verificatieverzoek
            print(f'E-mailverificatie verzoek: {request.url}')

        return response

# Voeg middleware toe aan app
app.add_middleware(
    EmailVerificationMiddleware,
    service=get_verification_service()
)

from fastapi import BackgroundTasks

@router.post('/verify-async')
async def verify_email_async(
    request: EmailVerificationRequest,
    background_tasks: BackgroundTasks,
    service: EmailVerificationService = Depends(get_verification_service)
):
    """Verifieer e-mail asynchroon."""

    async def verify_task(email: str):
        """Achtergrondtaak voor e-mailverificatie."""
        try:
            result = await service.verify_email(email)
            # Sla resultaat op in database
            await store_verification_result(email, result)
        except Exception as e:
            print(f'Achtergrond verificatie mislukt: {str(e)}')

    background_tasks.add_task(verify_task, request.email)

    return {
        'message': 'Verificatie gestart',
        'email': request.email
    }

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

class CacheService:
    """Service voor caching van e-mailverificatie resultaten."""

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

    async def connect(self):
        """Verbind met Redis."""
        self.redis = await redis.from_url(self.redis_url)

    async def disconnect(self):
        """Verbreek verbinding met Redis."""
        if self.redis:
            await self.redis.close()

    async def get(self, key: str) -> Optional[Any]:
        """Haal waarde op uit 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):
        """Stel waarde in cache met TTL."""
        if not self.redis:
            return

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

    async def delete(self, key: str):
        """Verwijder waarde uit cache."""
        if not self.redis:
            return

        await self.redis.delete(key)

    async def clear(self):
        """Wis alle cache."""
        if not self.redis:
            return

        await self.redis.flushdb()

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

# Startup en shutdown events
@app.on_event('startup')
async def startup():
    """Verbind met cache bij opstarten."""
    cache_service = get_cache_service()
    await cache_service.connect()

@app.on_event('shutdown')
async def shutdown():
    """Verbreek verbinding met cache bij afsluiten."""
    cache_service = get_cache_service()
    await cache_service.disconnect()

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 e-mailverificatie endpoint."""
    return await service.verify_email(email_request.email)

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 voor e-mailverificatie resultaten."""

    __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):
        """Haal async database session op."""
        async with self.SessionLocal() as session:
            yield session

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

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

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

client = TestClient(app)

@pytest.fixture
def mock_service():
    """Mock e-mailverificatie 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 enkel e-mailverificatie."""
    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 ongeldige e-mailverificatie."""
    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 e-mailverificatie."""
    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

FROM python:3.11-slim

WORKDIR /app

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

# Kopieer applicatie
COPY . .

# Expose poort
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')"

# Start applicatie
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

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