Flaskでメールアドレスを検証する方法

Flaskでメールアドレスをバリデーションするにはチェックがひとつではなく3つ必要ですが、Flask自体にはそのどれも含まれていません。Flaskにはフォームの仕組みがなく、あるのはエコシステムだけだからです。フォーム処理にはWTForms（通常はFlask-WTF経由）、そのWTFormsが内部で利用するemail-validatorパッケージがあります。この2つで、検証の3つの問いのうち最初の2つ、つまりアドレスの形式が正しいか、そのドメインがメールを受信できるかには答えられます。しかし3つ目の、メールボックスが実在するかどうかには答えられません。実際の検証は、構文チェック、DNSルックアップ、SMTPによるメールボックスの確認という3層で行います。このガイドでは、各層をWTFormsのバリデーターとして実装し、Flask-WTFのフォームにも通常のJSONルートにも組み込める形にします。そのうえで、検証APIにどこから任せるかも示します。Pythonガイドの続きにあたるため、そちらで説明したDNSとSMTPの考え方はそのまま当てはまります。

結論#

構文にはWTFormsのEmail()バリデーター、MXルックアップにはそのオプションcheck_deliverability=True（またはdnspython）、SMTPによるメールボックスの確認には検証APIを使います。コストの低い順に並べ、各層が次の層の前に処理を打ち切れるようにします。メールボックスの確認のために、FlaskのプロセスからSMTP接続を開くのはやめてください。アウトバウンドのポート25はほとんどのホストでブロックされています。また、結果は送信元IPのレピュテーションやグレーリスティングに左右されるため、自前で再現するのは得策ではありません。各層は、前の層より低いコストでアドレスを除外できますが、アドレスを有効と確定できるのはAPIだけです。この流れに馴染みがない場合は、まずメールアドレス検証とはで用語を確認してください。

レイヤー1：WTFormsのEmailバリデーターによる形式の検証#

レイヤー1はWTFormsが担います。自前で実装しているわけではなく、wtforms.validators.Emailはemail-validatorパッケージに処理を任せています。そのため、フィールドにEmail()バリデーターを宣言するだけで、正規表現ではなくRFCに基づく本格的な構文チェックが手に入ります。

from flask_wtf import FlaskForm
from wtforms import StringField
from wtforms.validators import DataRequired, Email

class SignupForm(FlaskForm):
    email = StringField("Email", validators=[DataRequired(), Email()])

Email()バリデーターを使うにはパッケージのインストールが必要です（pip install email-validator、またはpip install wtforms[email]）。未インストールの場合、WTFormsは検証時にその旨を伝えるエラーを発生させます。内部ではemail_validator.validate_emailを呼び出しており、フォームの外やサービス層でも単独で使えば、同じ判定を真偽値として得られます。

from email_validator import validate_email, EmailNotValidError

def is_valid_syntax(email: str) -> bool:
    try:
        validate_email(email, check_deliverability=False)
        return True
    except EmailNotValidError:
        return False

このチェックで何が得られるのかを、正確に把握しておいてください。check_deliverability=Falseのとき、確認するのは形式だけです。DNSの解決もソケットの接続も行わず、example.comが存在するかどうかも分かりません。definitely-fake@gmail.comは通ります。info@company-that-folded.comも通ります。typo@gmial.comも通ります。3つとも配信不能ですが、文字列を読むだけのバリデーターには見抜けません。正規表現によるメールアドレス検証が失敗するのと同じ理由です。構文と到達可否は別の問題であり、前者は文字列についての事実、後者はインターネットについての事実だからです。

レイヤー2：ドメインはメールを受け付けるか#

MXレコードのないドメインは誰宛てのメールも受け付けられません。そのため、DNSルックアップを1回行うだけで、存在しないドメイン、社名のスペルミス、架空のTLDを除外できます。ここでFlaskのスタックは、Djangoが標準では行わないことをしてくれます。email-validatorがルックアップを代行できるのです。check_deliverabilityを有効にすると、dnspythonでドメインのMXレコード（なければA/AAAAにフォールバック）を解決し、ドメインがメールを受信できない場合はEmailNotValidErrorのサブクラスであるEmailUndeliverableErrorを発生させます。フォームでは、すでに使っているバリデーターにオプションを1つ付けるだけです。

email = StringField("Email", validators=[DataRequired(), Email(check_deliverability=True)])

このフラグ1つで、レイヤー1と2が1つのバリデーターにまとまります。単独で呼び出した場合は、ドメインに問題がなければ正規化されたアドレスのオブジェクトが返ります。

from email_validator import validate_email, EmailNotValidError

def has_mail_route(email: str) -> bool:
    try:
        validate_email(email, check_deliverability=True)  # syntax + live MX lookup
        return True
    except EmailNotValidError:
        return False

結果をキャッシュしたい、独自のリゾルバーを指定したい、タイムアウトを短くしたいといった理由で、DNSチェックを自分で行いたい場合は、dnspythonを直接使います。

import dns.resolver  # pip install dnspython

def domain_has_mx(domain: str) -> bool:
    try:
        return len(dns.resolver.resolve(domain, "MX")) > 0
    except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.NoNameservers):
        return False

レイヤー3：メールボックスは実在するか#

レイヤー1と2でできるのは、アドレスを除外することだけです。ドメインに正しいMXレコードがあっても、手元のアドレスにメールボックスがあるとは限りません。noreply-9f2x@gmail.comは構文が正しく、メール経路も有効ですが、メールボックスは作成されたことがありません。特定のメールボックスを確認するには、SMTPの配信時のやり取りが必要です。メールホストに接続し、RCPT TOを発行して応答を読み取り、何も送信せずに切断します。

原理的には、Flaskからsmtplibでこれをスクリプト化できます。しかし、アプリケーションサーバーから実行すべきではありません。ほとんどのクラウドホストはアウトバウンドのポート25をブロックしており、応答は接続元IPのレピュテーションに左右されます。さらに、受信サーバーは見慣れない送信者に対してグレーリスティングやレート制限を行います。そのため、ローカルのテストでは動いたプローブが、本番では気づかないうちに失敗したり、接続元IPがブラックリストに載ったりします。キャッチオールドメインを含む全体のパイプラインについては、メールアドレス検証の仕組みで解説しています。ここは外部に任せる価値のある層です。

Qualisendの検証エンドポイントは、構文、DNS、SMTPによるメールボックスの確認というパイプライン全体を、レピュテーションが管理されたインフラから実行し、判定結果を返します。requestsで呼び出し、キーは環境変数から読み込みます。

import os
import requests  # pip install requests

QUALISEND_VERIFY_URL = "https://app.qualisend.com/api/v1/verify"

def verify(email: str) -> dict:
    res = requests.post(
        QUALISEND_VERIFY_URL,
        headers={"Authorization": f"Bearer {os.environ['QUALISEND_API_KEY']}"},
        json={"email": email},
        timeout=10,
    )
    res.raise_for_status()
    return res.json()["result"]  # {"status", "score", "reason", "sub_flags", "smtp"}

レスポンスは{ "result": { ... } }というエンベロープ形式です。result["status"]（deliverable、risky、undeliverable、unknownのいずれか）を、score、reason、役割アドレスや使い捨てアドレスなどの特徴を示すsub_flagsとあわせて読み取ります。フィールドの正確な形式はAPIリファレンスで確認してください。判断に必要なのはstatusだけです。undeliverableを拒否するのが安全な基本方針です。riskyも拒否する、あるいはscoreを保存して通す、といった選択もあり、これは運用ポリシーの問題です。収集時点でどこまで厳格にするかは、サーバーレスのサインアップガイドで解説しています。

このstatusがどこまで確定したものかは、result["smtp"]によって決まります。cachedまたはskippedなら、判定はすでに確定しています。pendingは、SMTPによるメールボックスの確認がまだ実行中であることを示します。この場合、statusはローカルチェックの結果を反映したもので、reasonはnullのままです。確定した判定はGET /jobs/{job_id}で取得できます。job_idは同じレスポンスの最上位（resultの隣）にあります。上記のverify()はresultだけを返すため、smtpがpendingのときはres.json()["job_id"]も保持してください。このポーリングはバックグラウンドタスクで行い、リクエストの中では行わないでください。また、同じアドレスに対して/verifyを再度呼び出すのもやめてください。サインアップ時は、即時のstatusに基づいて処理し、完了したジョブの結果がundeliverableだった場合は、そのバックグラウンドタスクでアカウントにフラグを立てます（?include=resultsを付けてresults[0].statusを読み取ります。ジョブ自体のstatusはcompletedを示すだけです）。

3つの層を組み合わせてFlaskでメールアドレスを検証する#

Flask-WTFでフィールド単位のチェックを追加する定番の方法は、フォームにインラインのvalidate_<fieldname>メソッドを定義することです。WTFormsはこのメソッドをフィールドのバリデーターチェーンの末尾に追加し、リストにあるバリデーターの後に実行します。そのため、このメソッドが呼ばれる時点では、Email(check_deliverability=True)による構文とMXルックアップの判定がすでに済んでいます。field.errorsでガードを入れておけば、コストの低い2つの層を通過したアドレスに対してだけAPIの往復が発生します。

from flask_wtf import FlaskForm
from wtforms import StringField
from wtforms.validators import DataRequired, Email, ValidationError
import requests

class SignupForm(FlaskForm):
    email = StringField("Email", validators=[
        DataRequired(),
        Email(check_deliverability=True),   # layers 1 + 2
    ])

    def validate_email(self, field):        # layer 3 — runs after the list above
        if field.errors:                    # syntax or MX already failed
            return                          # don't spend an API credit
        try:
            result = verify(field.data)
        except requests.RequestException:   # timeout, 402, 429, 5xx: fail open
            return                          # let the signup through, re-verify later
        if result["status"] == "undeliverable":
            raise ValidationError("We couldn't confirm a mailbox at this address.")

肝心なのは、このfield.errorsのガードです。形式が不正なアドレスや存在しないドメインは、インラインメソッドが実行される前にチェーン内でエラーとして記録されます。ガードがあれば、通過する見込みのないアドレスで検証APIが呼ばれることはありません。検証の順序を誤ったり、ガードを省略したりすると、入力ミスのたびにクレジットを消費します。ビューでは、validate_on_submitがハンドラー本体の実行前にパイプライン全体を実行します。

@app.route("/signup", methods=["POST"])
def signup():
    form = SignupForm()
    if form.validate_on_submit():
        # every layer passed — safe to persist
        create_account(form.email.data)
        return redirect(url_for("welcome"))
    return render_template("signup.html", form=form), 400

あらかじめ決めておくべきことがあります。API呼び出し自体が失敗したときの動作です。validate_email内でのネットワークのタイムアウトや、raise_for_status()が402（クレジット不足）、429（レート制限）、5xxに対して発生させるHTTPErrorで、リクエストを500にしたり、実在する顧客に対処しようのないエラーを見せたりしてはいけません。requests.RequestExceptionはValidationErrorとは別に捕捉し、サービスに到達できない場合はundeliverableではなくunknownとして扱ってください。一時的な障害で登録をブロックするのではなく、サインアップを通したうえで、後からアドレスを再検証します。構文とMXの層はすでに実行済みなので、緩めるのはネットワークに依存する層だけです。

すべてのFlaskアプリがWTFormsを使うわけではありません。JSON APIでも、同じ3つの層をルートの中にそのまま書けます。この場合、ValidationErrorを発生させる代わりに、不正なアドレスには400を返します。

from flask import Flask, request, jsonify
from email_validator import validate_email, EmailNotValidError
import requests

app = Flask(__name__)

@app.post("/api/signup")
def api_signup():
    email = (request.get_json(silent=True) or {}).get("email", "")
    try:
        info = validate_email(email, check_deliverability=True)  # layers 1 + 2
    except EmailNotValidError as exc:
        return jsonify(error=str(exc)), 400

    try:
        status = verify(info.normalized)["status"]                # layer 3
    except requests.RequestException:                             # timeout, 402, 429, 5xx
        status = "unknown"                                        # fail open, re-verify later
    if status == "undeliverable":
        return jsonify(error="Email address appears undeliverable."), 400

    # ... create the account
    return jsonify(ok=True), 201

info.normalizedはemail-validatorが返す正規化済みのアドレスで、保存すべき値です。この3層の構成は、このガイドのNode.js版やPython版と同じです。検証を機能させるのは、フレームワークではなく層構造です。レイヤー3を支える検証サービスを選ぶ際は、APIの比較で選択肢を並べて確認できます。

SMTPの層を追加する準備はできましたか。Email()を通過したアドレスを無料メールチェッカーに入力すると、WTFormsでは有効とされた文字列がundeliverableと判定される様子を確認できます。そのうえで、コピー&ペーストで使えるコード例のあるAPIリファレンスを参考に、同じ判定結果をフォームに組み込んでください。