Contact Export → Bulk Persona Creation

Export SendGrid marketing contacts and batch-create Mavera personas from subscriber segments at scale

Scenario

Your SendGrid marketing contacts hold thousands of subscribers with custom fields — company size, industry, role, signup source, engagement score. Creating personas one by one won’t scale. This job triggers a SendGrid contact export, downloads the CSV, segments contacts by meaningful clusters, then batch-creates Mavera personas for each segment. The result is a persona library that mirrors your actual subscriber base, ready for focus groups and content generation.

Flow: SendGrid POST /v3/marketing/contacts/exports → Poll → Download CSV → Segment by custom fields → Mavera POST /api/v1/personas (batch) → Persona library at scale

Architecture

Code

Python

import os, csv, io, requests, time
from collections import defaultdict

SG = os.environ["SENDGRID_API_KEY"]
MV = os.environ["MAVERA_API_KEY"]
SG_BASE = "https://api.sendgrid.com/v3"
MB = "https://app.mavera.io/api/v1"
SG_H = {"Authorization": f"Bearer {SG}", "Content-Type": "application/json"}
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}

# 1. Trigger export
export = requests.post(f"{SG_BASE}/marketing/contacts/exports",
    headers=SG_H, json={
        "list_ids": [],
        "segment_ids": [],
        "file_type": "csv",
        "max_file_size": 5000,
    }).json()

export_id = export.get("id")
print(f"Export triggered: {export_id}")

# 2. Poll until ready
download_url = None
for attempt in range(30):
    time.sleep(10)
    status = requests.get(f"{SG_BASE}/marketing/contacts/exports/{export_id}",
        headers=SG_H).json()
    state = status.get("status", "pending")
    print(f"  Export status: {state} (attempt {attempt + 1})")

    if state == "ready":
        urls = status.get("urls", [])
        if urls:
            download_url = urls[0]
        break
    elif state == "failure":
        print(f"Export failed: {status.get('message', 'unknown')}")
        exit()

if not download_url:
    print("Export timed out")
    exit()

# 3. Download and parse CSV
csv_data = requests.get(download_url).text
reader = csv.DictReader(io.StringIO(csv_data))
contacts = list(reader)
print(f"Downloaded {len(contacts)} contacts")

# 4. Cluster by industry × role
clusters = defaultdict(list)
for contact in contacts:
    industry = (contact.get("industry") or contact.get("custom_industry") or "unknown").strip().lower()
    role = (contact.get("job_title") or contact.get("custom_role") or "unknown").strip().lower()
    engagement = "high" if int(contact.get("engagement_score", 0) or 0) > 70 else \
                 "medium" if int(contact.get("engagement_score", 0) or 0) > 30 else "low"

    key = f"{industry}|{role}|{engagement}"
    clusters[key].append(contact)

# 5. Create personas for significant clusters
significant = {k: v for k, v in clusters.items() if len(v) >= 10}
print(f"Significant clusters (10+): {len(significant)}")

personas = []
for key, members in sorted(significant.items(), key=lambda x: -len(x[1]))[:20]:
    industry, role, engagement = key.split("|")
    emails_domains = list({
        m.get("email", "").split("@")[-1]
        for m in members if "@" in m.get("email", "")
    })[:5]
    sources = list({m.get("signup_source", "") for m in members if m.get("signup_source")})[:3]

    r = requests.post(f"{MB}/personas", headers=MV_H, json={
        "name": f"SG: {role.title()} / {industry.title()} / {engagement.title()}",
        "description": (
            f"SendGrid segment. Role: {role}. Industry: {industry}. "
            f"Engagement: {engagement}. N={len(members)}. "
            f"Top domains: {', '.join(emails_domains[:3])}. "
            f"Sources: {', '.join(sources)}."
        ),
        "demographic": {"job_titles": [role], "industries": [industry]},
        "psychographic": {"engagement_level": engagement, "signup_sources": sources},
    })
    r.raise_for_status()
    personas.append({"cluster": key, "id": r.json()["id"], "n": len(members)})
    print(f"  {role.title()} / {industry.title()} / {engagement.title()}: "
          f"{r.json()['id']} ({len(members)})")
    time.sleep(0.3)

print(f"\nCreated {len(personas)} personas from {len(contacts)} contacts")

Example Output

Export triggered: exp_8f3a2b1c
  Export status: pending (attempt 1)
  Export status: pending (attempt 2)
  Export status: ready (attempt 3)
Downloaded 8,432 contacts
Significant clusters (10+): 14

  Marketing Manager / Saas / High: per_sg_b1c2 (342)
  Director / Technology / Medium: per_sg_d3e4 (218)
  Founder / Ecommerce / High: per_sg_f5g6 (187)
  Vp Marketing / Fintech / Medium: per_sg_h7i8 (93)
  Product Manager / Healthcare / Low: per_sg_j9k0 (64)
  ...

Created 14 personas from 8,432 contacts

Error Handling

Export takes too long

Large contact lists (100K+) can take 5-10 minutes to export. The code polls for 5 minutes (30 attempts × 10s). Increase for very large lists or use segment-specific exports.

CSV parsing edge cases

Contact fields may contain commas within quoted strings. The JavaScript parser uses regex matching for quoted fields. For production, use a proper CSV parser like csv-parse (Node) or Python’s built-in csv module.

Custom field names

Custom fields in SendGrid use the names you defined. Check your fields at GET /v3/marketing/field_definitions. Common patterns: custom_industry, custom_role, or the exact name you set.

Export rate limit

You can only have one active export at a time. If a previous export is in progress, the API returns 429. Wait for it to complete before triggering a new one.