Email Verification in Python

pip install mailbeam

# Or with Poetry:
poetry add mailbeam

# Or add to requirements.txt:
echo "mailbeam>=1.0" >> requirements.txt
pip install -r requirements.txt

# .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 os
import mailbeam
from fastapi import HTTPException

mb = mailbeam.Client(api_key=os.environ["MAILBEAM_KEY"])


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 mb.verify(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",
                },
            )
    except mailbeam.APIError as exc:
        # Log but don't block — fail open on Mailbeam errors
        import logging
        logging.error("Mailbeam verification failed: %s", exc)

    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 os
import mailbeam
import logging

mb = mailbeam.Client(api_key=os.environ["MAILBEAM_KEY"])


def verify_email_sync(email: str) -> tuple[bool, str | None]:
    """
    Returns (is_valid, reason_code).
    Fails open (returns True) on API errors.
    """
    try:
        result = mb.verify_sync(email)
        if not result.valid or result.score < 60:
            return False, result.reason or "invalid_email"
        return True, None
    except mailbeam.APIError as exc:
        logging.error("Mailbeam verification failed: %s", exc)
        return True, None  # fail open


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

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 mb.verify(normalized)
        _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},
            )
    except mailbeam.APIError:
        pass  # fail open

    return email