郵件 API 開發者指南

串接範例

每個範例分配一個信箱，每 5 秒查信一次、最多 60 秒，讀取第一封新郵件詳情，並在結束時解除信箱。

Python 3 · 僅使用標準函式庫，無需安裝套件。

python3 dotmail.py

dotmail.py

# Python 3 · run on your server, keep the key private.
import hashlib
import hmac
import json
import os
import secrets
import time
import urllib.request
from urllib.parse import quote, urlencode

BASE = os.getenv("DOTMAIL_BASE_URL", "https://dotmail.cc").rstrip("/")
KEY = os.environ["DOTMAIL_API_KEY"]


def call(method, path, payload=None, query=""):
    body = json.dumps(payload, separators=(",", ":")).encode() if payload is not None else b""
    stamp, nonce = str(int(time.time())), secrets.token_hex(16)
    canonical = "\n".join([method, path, query, stamp, nonce, hashlib.sha256(body).hexdigest()])
    signature = hmac.new(KEY.encode(), canonical.encode(), hashlib.sha256).hexdigest()
    req = urllib.request.Request(BASE + path + ("?" + query if query else ""),
        data=body if payload is not None else None, method=method, headers={
        "X-Api-Key": KEY, "X-Timestamp": stamp, "X-Nonce": nonce,
        "X-Signature": signature, "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=30) as response:
        result = json.load(response)
        if result.get("code") != 200:
            raise RuntimeError(str(result))
        return result["data"]


def main():
    # Reuse this ID if allocation needs retrying; do not create a new charge.
    request_id = os.getenv("DOTMAIL_REQUEST_ID") or secrets.token_hex(16)
    print("request_id:", request_id)
    inbox = call("POST", "/api/v1/external/emails/allocate", {
        "site": os.getenv("DOTMAIL_SITE", "example.test"), "request_id": request_id})
    path = "/api/v1/external/emails/" + quote(inbox["email"], safe="") + "/messages"
    query = urlencode({"allocation_id": inbox["allocation_id"]})
    try:
        print(inbox["email"], inbox["allocation_id"])
        for _ in range(12):
            messages = call("GET", path, query=query)
            if messages["items"]:
                message_id = messages["items"][0]["id"]
                detail = call("GET", path + "/" + str(message_id), query=query)
                print(json.dumps(detail, ensure_ascii=False))
                break
            time.sleep(5)
        else:
            print("No new mail within 60 seconds.")
    finally:
        # Early release is free; it does not refund the allocation charge.
        try:
            call("POST", "/api/v1/external/emails/release", {
                "email": inbox["email"], "allocation_id": inbox["allocation_id"]})
        except Exception as error:
            print("Release failed; retry with the same allocation_id:", error)


if __name__ == "__main__":
    main()