SendGrid Bounces & Blocks: Handle Undeliverable Emails & Protect Reputation

# Hard Bounce: Permanent failure (address doesn't exist)
# Soft Bounce: Temporary failure (mailbox full, server busy)
# Block: ISP rejected (spam, blacklist, rate limit)

# SendGrid categorizes in webhook events:
# {
#   "event": "bounce",
#   "email": "alice@example.com",
#   "bounce_type": "permanent",  # or "transient"
#   "reason": "550 5.1.1 The email account that you tried to reach does not exist"
# }

def classify_bounce(event):
    bounce_type = event.get("bounce_type")
    reason = event.get("reason", "")

    if bounce_type == "permanent":
        print(f"Hard bounce: {event['email']} — remove from list")
        return "remove"
    elif bounce_type == "transient":
        print(f"Soft bounce: {event['email']} — retry later")
        return "retry"
    elif bounce_type == "block":
        print(f"Block: {event['email']} — investigate reason")
        return "investigate"

from flask import Flask, request, jsonify

app = Flask(__name__)
bounced_emails = set()

@app.route("/sendgrid/bounces", methods=["POST"])
def handle_bounces():
    events = request.get_json()

    for event in events:
        if event["event"] in ("bounce", "block", "dropped"):
            email = event["email"]
            event_type = event["event"]
            reason = event.get("reason", "No reason")

            print(f"[{event_type}] {email}: {reason[:100]}")

            if event_type == "bounce" and event.get("bounce_type") == "permanent":
                bounced_emails.add(email)
                remove_from_active_list(email)
                log_bounce(email, reason)

    return jsonify({"status": "ok"}), 200

def remove_from_active_list(email):
    print(f"Removed {email} from active send list")

def log_bounce(email, reason):
    print(f"Logged bounce: {email} — {reason[:50]}...")

def get_bounce_list():
    response = sg.client.suppression.bounces.get()
    bounces = response.to_dict()
    print(f"Total bounces: {len(bounces)}")
    for b in bounces[:5]:
        print(f"  {b['email']} — {b.get('reason', 'N/A')[:60]}")
    return bounces

def delete_bounce(email):
    response = sg.client.suppression.bounces._(email).delete()
    print(f"Removed {email} from bounces: {response.status_code}")

def clear_all_bounces():
    response = sg.client.suppression.bounces.delete()
    print(f"Cleared all bounces: {response.status_code}")

# Get current bounces
get_bounce_list()
# Expected output:
# Total bounces: 15
#   alice@old.com — 550 5.1.1 The email account that you tried to reach does not exist
#   bob@invalid.com — 550 5.1.1 User unknown

from datetime import datetime, timedelta

def check_bounce_rate(days=7):
    now = datetime.now()
    start = now - timedelta(days=days)

    params = {
        "start_date": start.strftime("%Y-%m-%d"),
        "end_date": now.strftime("%Y-%m-%d"),
        "aggregated_by": "day"
    }

    response = sg.client.stats.get(query_params=params)
    stats = response.to_dict()

    total_sent = 0
    total_bounces = 0

    for day in stats:
        metrics = day.get("stats", [{}])[0].get("metrics", {})
        total_sent += metrics.get("requests", 0)
        total_bounces += metrics.get("bounces", 0)

    bounce_rate = (total_bounces / max(total_sent, 1)) * 100

    print(f"Bounce rate (last {days} days): {bounce_rate:.2f}%")
    print(f"  Sent: {total_sent}")
    print(f"  Bounces: {total_bounces}")

    if bounce_rate > 5:
        print("CRITICAL: Bounce rate exceeds 5% — clean your list immediately!")
    elif bounce_rate > 3:
        print("WARNING: Bounce rate above 3% — review list hygiene")
    else:
        print("OK: Bounce rate is healthy")

check_bounce_rate()
# Expected output: Bounce rate (last 7 days): 0.8%
#                  OK: Bounce rate is healthy