Send Email from Google Cloud Functions (Python)

import json, os, urllib.request
payload = {
    "from": {"email": os.environ["AHASEND_FROM"]},
    "recipients": [{"email": "recipient@example.com"}],
    "subject": "My first sandbox email",
    "text_content": "Hello from AhaSend",
    "sandbox": True,
}
url = f"https://api.ahasend.com/v2/accounts/{os.environ['AHASEND_ACCOUNT_ID']}/messages"
request = urllib.request.Request(url, json.dumps(payload).encode(), headers={
    "Authorization": f"Bearer {os.environ['AHASEND_API_KEY']}",
    "Content-Type": "application/json",
})
with urllib.request.urlopen(request, timeout=10) as response:
    statuses = [item["status"] for item in json.load(response)["data"]]
if not statuses or any(s not in ("queued", "scheduled") for s in statuses):
    raise RuntimeError("A recipient was rejected")
print({"statuses": statuses})

import hmac, json, os, re, urllib.error, urllib.request

def send_email(body, authorization):
    token = os.environ.get("SEND_TOKEN", "")
    if not token:
        return 500, {"error": "Missing server configuration"}
    if not hmac.compare_digest(authorization.encode(), ("Bearer " + token).encode()):
        return 401, {"error": "Unauthorized"}
    event_id = body.get("event_id") if isinstance(body, dict) else None
    if not isinstance(event_id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{8,64}", event_id):
        return 400, {"error": "event_id must be 8-64 letters, digits, _ or -"}
    mode = os.environ.get("AHASEND_DELIVERY_MODE", "sandbox")
    required = ("AHASEND_API_KEY", "AHASEND_ACCOUNT_ID", "AHASEND_FROM", "AHASEND_TO")
    if mode not in ("sandbox", "live") or any(not os.environ.get(k) for k in required):
        return 500, {"error": "Missing or invalid server configuration"}
    payload = {
        "from": {"email": os.environ["AHASEND_FROM"]},
        "recipients": [{"email": os.environ["AHASEND_TO"]}],
        "subject": "Welcome",
        "text_content": "Your account is ready.",
        "sandbox": mode == "sandbox",
    }
    url = f"https://api.ahasend.com/v2/accounts/{os.environ['AHASEND_ACCOUNT_ID']}/messages"
    request = urllib.request.Request(url, json.dumps(payload).encode(), headers={
        "Authorization": f"Bearer {os.environ['AHASEND_API_KEY']}",
        "Content-Type": "application/json",
        "Idempotency-Key": f"{mode}-welcome-{event_id}",
    })
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            result = json.load(response)
        statuses = [item["status"] for item in result["data"]]
        if not statuses or any(s not in ("queued", "scheduled") for s in statuses):
            return 422, {"error": "A recipient was rejected"}
        return 202, {"statuses": statuses}
    except (urllib.error.URLError, TimeoutError, ValueError, KeyError, TypeError):
        return 502, {"error": "Email request failed"}

import functions_framework
from send_email import send_email

@functions_framework.http
def send_email_http(request):
    if request.method != "POST":
        return {"error": "Use POST"}, 405
    if len(request.get_data()) > 16384:
        return {"error": "Request too large"}, 413
    if not request.is_json:
        return {"error": "Use application/json"}, 415
    body = request.get_json(silent=True)
    if body is None:
        return {"error": "Invalid JSON"}, 400
    status, result = send_email(body, request.headers.get("authorization", ""))
    return result, status