Send Email from Azure 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 json
import azure.functions as func
from send_email import send_email
app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

@app.route(route="sendEmail", methods=["POST"])
def send_email_http(req: func.HttpRequest) -> func.HttpResponse:
    raw = req.get_body()
    if len(raw) > 16384:
        return func.HttpResponse(status_code=413)
    try:
        body = json.loads(raw)
    except (ValueError, UnicodeError):
        return func.HttpResponse("Invalid JSON", status_code=400)
    status, result = send_email(body, req.headers.get("authorization", ""))
    return func.HttpResponse(json.dumps(result), status_code=status, mimetype="application/json")

{"version":"2.0"}