Email Verification in Python

pip install httpx        # or use requests, if that is already in the project

# app/mailbeam.py
import os

import httpx

ENDPOINT = "https://api.mailbeam.dev/v1/verify"
HEADERS = {"Authorization": f"Bearer {os.environ['MAILBEAM_KEY']}"}


class MailbeamError(Exception):
    """Anything that stopped us getting a verdict."""


async def verify_email(email: str, timeout: float = 3.0) -> dict:
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.post(ENDPOINT, json={"email": email}, headers=HEADERS)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as exc:
        raise MailbeamError(str(exc)) from exc


def verify_email_sync(email: str, timeout: float = 3.0) -> dict:
    try:
        response = httpx.post(ENDPOINT, json={"email": email}, headers=HEADERS, timeout=timeout)
        response.raise_for_status()
        return response.json()
    except httpx.HTTPError as exc:
        raise MailbeamError(str(exc)) from exc

# .env
MAILBEAM_KEY=mb_live_xxxxxxxxxxxxxxxxxxxx

import os
from dotenv import load_dotenv

load_dotenv()
MAILBEAM_KEY = os.environ["MAILBEAM_KEY"]

# app/dependencies.py
import logging

from fastapi import HTTPException

from app.mailbeam import MailbeamError, verify_email


async def validate_email(email: str) -> str:
    """
    FastAPI dependency that verifies an email address.
    Raises HTTP 422 if the email is invalid or low-quality.
    Fails open (passes through) on Mailbeam API errors.
    """
    try:
        result = await verify_email(email)
    except MailbeamError as exc:
        # Log but don't block — our outage is not the user's problem.
        logging.error("Mailbeam verification failed: %s", exc)
        return email

    if not result["valid"] or result["score"] < 60:
        raise HTTPException(
            status_code=422,
            detail={
                "error": "Please provide a valid email address.",
                "code": result["reason"] or "invalid_email",
            },
        )

    return email

# app/routes/auth.py
from fastapi import APIRouter, Depends
from pydantic import BaseModel, EmailStr
from typing import Annotated
from app.dependencies import validate_email

router = APIRouter()


class SignupRequest(BaseModel):
    email: EmailStr
    password: str


@router.post("/signup", status_code=201)
async def signup(
    request: SignupRequest,
    # The dependency runs BEFORE the handler and raises 422 if invalid
    verified_email: Annotated[str, Depends(validate_email)],
):
    user = await create_user(email=verified_email, password=request.password)
    return {"user": user}

# app/main.py
from fastapi import FastAPI
from app.routes.auth import router

app = FastAPI()
app.include_router(router, prefix="/api/auth")

# app/utils/email.py
import logging

from app.mailbeam import MailbeamError, verify_email_sync


def check_email(email: str) -> tuple[bool, str | None]:
    """
    Returns (is_valid, reason_code).
    Fails open (returns True) on API errors.
    """
    try:
        result = verify_email_sync(email)
    except MailbeamError as exc:
        logging.error("Mailbeam verification failed: %s", exc)
        return True, None  # fail open

    if not result["valid"] or result["score"] < 60:
        return False, result["reason"] or "invalid_email"
    return True, None


# app/routes/auth.py (Flask)
from flask import Blueprint, request, jsonify
from app.utils.email import check_email

bp = Blueprint("auth", __name__)


@bp.post("/api/auth/signup")
def signup():
    data = request.get_json()
    email = data.get("email", "")

    is_valid, reason = verify_email_sync(email)
    if not is_valid:
        return jsonify({"error": "Please provide a valid email.", "code": reason}), 422

    user = create_user(email=email, password=data.get("password"))
    return jsonify({"user": user}), 201

# tests/test_signup.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)


def test_signup_with_valid_email():
    # Uses Mailbeam's test domain — always returns valid, doesn't count against quota
    response = client.post(
        "/api/auth/signup",
        json={"email": "user@valid.mailbeam-test.dev", "password": "pass1234"},
    )
    assert response.status_code == 201


def test_signup_with_invalid_email():
    response = client.post(
        "/api/auth/signup",
        json={"email": "user@invalid.mailbeam-test.dev", "password": "pass1234"},
    )
    assert response.status_code == 422
    assert response.json()["detail"]["code"] is not None


def test_signup_with_disposable_email():
    response = client.post(
        "/api/auth/signup",
        json={"email": "temp@disposable.mailbeam-test.dev", "password": "pass1234"},
    )
    assert response.status_code == 422

# app/dependencies.py
from functools import lru_cache
import asyncio

# Simple in-process TTL cache using a dict (use Redis in production)
_cache: dict[str, tuple] = {}

async def validate_email(email: str) -> str:
    normalized = email.lower().strip()

    if normalized in _cache:
        valid, score, reason = _cache[normalized]
        if not valid or score < 60:
            raise HTTPException(
                status_code=422,
                detail={"error": "Please provide a valid email.", "code": reason},
            )
        return email

    try:
        result = await verify_email(normalized)
    except MailbeamError:
        return email  # fail open

    _cache[normalized] = (result["valid"], result["score"], result["reason"])

    if not result["valid"] or result["score"] < 60:
        raise HTTPException(
            status_code=422,
            detail={"error": "Please provide a valid email.", "code": result["reason"]},
        )

    return email