Hubspot Bulk Migration

def relink_associations(source_assocs, from_id_map, to_id_map,
                         from_type, to_type, type_id, batch_size=100):
    """Re-create associations using v4 API. Returns {linked, skipped, errors}."""
    url = f"{BASE}/crm/v4/associations/{from_type}/{to_type}/batch/create"
    linked = skipped = 0
    errors = []
    batches = [source_assocs[i:i+batch_size] for i in range(0, len(source_assocs), batch_size)]
    for batch in batches:
        inputs = []
        for a in batch:
            fhs = from_id_map.get(a["from_source_id"])
            ths = to_id_map.get(a["to_source_id"])
            if not fhs or not ths:
                skipped += 1; continue
            inputs.append({"from": {"id": fhs}, "to": {"id": ths},
                            "types": [{"associationCategory": "HUBSPOT_DEFINED", "associationTypeId": type_id}]})
        if not inputs: continue
        resp = requests.post(url, headers=HDRS, json={"inputs": inputs})
        if resp.status_code == 429:
            time.sleep(int(resp.headers.get("Retry-After", 10)))
            resp = requests.post(url, headers=HDRS, json={"inputs": inputs})
        if resp.status_code not in (200, 201, 207):
            errors.append({"status": resp.status_code, "body": resp.text[:200]})
        else:
            linked += len(inputs)
        time.sleep(0.12)
    return {"linked": linked, "skipped": skipped, "errors": errors}

# Standard association type IDs — full table in API_REFERENCE.md
CONTACT_TO_COMPANY, DEAL_TO_CONTACT, DEAL_TO_COMPANY = 1, 3, 5

def rollback_migration(id_map_path: str, object_type: str, dry_run: bool = True) -> dict:
    """Archive all records created during migration. Always dry_run=True first."""
    with open(id_map_path) as f:
        id_map = json.load(f)
    hubspot_ids = list(id_map.values())
    archived, errors = 0, []
    for i in range(0, len(hubspot_ids), 100):
        batch = hubspot_ids[i:i+100]
        if dry_run:
            print(f"Would archive: {batch[:3]}...")
            archived += len(batch); continue
        resp = requests.post(f"{BASE}/crm/v3/objects/{object_type}/batch/archive",
                             headers=HDRS, json={"inputs": [{"id": hid} for hid in batch]})
        if resp.status_code == 204:
            archived += len(batch)
        else:
            errors.append({"status": resp.status_code})
        time.sleep(0.12)
    return {"archived": archived, "errors": errors, "dry_run": dry_run}

# 1. Create ID properties
for obj in ["contacts", "companies", "deals"]:
    create_source_id_property(obj, "salesforce")

# 2. Import companies first
import csv
company_rows = [{"salesforce_id": r["Id"], "name": r["Name"],
                  "domain": r.get("Website","").replace("https://","").rstrip("/")}
                for r in csv.DictReader(open("sf_accounts.csv"))]
company_map, _ = batch_import(company_rows, "companies", "salesforce_id")
save_id_map(company_map, "company_id_map.json")

# 3. Import contacts (upsert by email for dedup)
contact_rows = [{"salesforce_id": r["Id"],
                  "email": r.get("Email","").strip().lower(),
                  "firstname": r.get("FirstName",""), "lastname": r.get("LastName","")}
                for r in csv.DictReader(open("sf_contacts.csv"))]
contact_map, _ = batch_import(contact_rows, "contacts", "salesforce_id")
save_id_map(contact_map, "contact_id_map.json")

# 4. Re-link contact → company
source_assocs = [{"from_source_id": r["Id"], "to_source_id": r["AccountId"]}
                 for r in csv.DictReader(open("sf_contacts.csv")) if r.get("AccountId")]
print(relink_associations(source_assocs, contact_map, company_map,
                           "contacts", "companies", CONTACT_TO_COMPANY))

curl -s "https://api.hubapi.com/crm/v3/objects/contacts?properties=email,salesforce_id&limit=5" \
  -H "Authorization: Bearer $HUBSPOT_ACCESS_TOKEN" | \
  jq '.results[] | {id, email: .properties.email, sf_id: .properties.salesforce_id}'

print(rollback_migration("contact_id_map.json", "contacts", dry_run=True))
# Confirm output, then:
# rollback_migration("contact_id_map.json", "contacts", dry_run=False)