War Story: Migrating 10k Leads from Salesforce to HubSpot 10.0 with Python 3.14

# salesforce_extractor.py
# Python 3.14 rc2+
# Extracts 10k leads from Salesforce using Bulk API 2.0 with retry logic
# and schema validation against HubSpot 10.0 target fields

import asyncio
import json
import logging
from datetime import datetime, timezone
from typing import List, Dict, Optional

import aiohttp
from simple_salesforce import AsyncSalesforce # 3.14-compatible build 2.1.1

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("migration.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# HubSpot 10.0 required fields for lead mapping (validated against schema v3.2)
HUBSPOT_REQUIRED_FIELDS = {"email", "firstname", "lastname", "company", "phone"}

class SalesforceExtractor:
    def __init__(self, sf_instance_url: str, sf_token: str, sf_client_id: str, sf_client_secret: str):
        self.sf = None
        self.sf_instance_url = sf_instance_url
        self.sf_token = sf_token
        self.sf_client_id = sf_client_id
        self.sf_client_secret = sf_client_secret
        self.extracted_leads: List[Dict] = []

    async def __aenter__(self):
        # Initialize async Salesforce client with 3.14's improved SSL context
        self.sf = await AsyncSalesforce(
            instance_url=self.sf_instance_url,
            token=self.sf_token,
            client_id=self.sf_client_id,
            client_secret=self.sf_client_secret,
            ssl_context=None # Use 3.14's default SSL with post-quantum crypto support
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.sf:
            await self.sf.close()

    async def extract_leads(self, query: str = "SELECT Id, Email, FirstName, LastName, Company, Phone, CreatedDate FROM Lead WHERE IsConverted = FALSE LIMIT 10000") -> List[Dict]:
        """Extract leads via Bulk API 2.0 with automatic retry for rate limits"""
        retry_count = 0
        max_retries = 5
        while retry_count <= max_retries:
            try:
                logger.info(f"Starting lead extraction with query: {query[:50]}...")
                # Use Bulk API 2.0 for 10k record sets (faster than REST)
                job = await self.sf.bulk2.Lead.query(query)
                await job.wait()
                result = await job.get_results()
                self.extracted_leads = await self._validate_and_normalize(result)
                logger.info(f"Extracted {len(self.extracted_leads)} valid leads")
                return self.extracted_leads
            except aiohttp.ClientResponseError as e:
                if e.status == 429: # Rate limit hit
                    retry_after = int(e.headers.get("Retry-After", 10))
                    logger.warning(f"Rate limit hit, retrying after {retry_after}s")
                    await asyncio.sleep(retry_after)
                    retry_count +=1
                else:
                    logger.error(f"Salesforce API error: {e}")
                    raise
            except Exception as e:
                logger.error(f"Unexpected extraction error: {e}")
                raise
        raise RuntimeError("Max retries exceeded for Salesforce extraction")

    async def _validate_and_normalize(self, raw_leads: List[Dict]) -> List[Dict]:
        """Normalize Salesforce fields to HubSpot 10.0 compatible schema"""
        normalized = []
        for lead in raw_leads:
            # Skip leads without required email (HubSpot 10.0 rejects these)
            if not lead.get("Email"):
                logger.warning(f"Skipping lead {lead.get('Id')} - no email")
                continue
            # Map Salesforce fields to HubSpot v3 field names
            normalized_lead = {
                "email": lead.get("Email", "").strip().lower(),
                "firstname": lead.get("FirstName", "").strip(),
                "lastname": lead.get("LastName", "").strip(),
                "company": lead.get("Company", "").strip(),
                "phone": lead.get("Phone", "").replace(" ", "").replace("-", ""),
                "salesforce_id": lead.get("Id"),
                "created_date": datetime.strptime(lead.get("CreatedDate", ""), "%Y-%m-%dT%H:%M:%S.%f%z").isoformat() if lead.get("CreatedDate") else None
            }
            # Validate required fields
            missing = HUBSPOT_REQUIRED_FIELDS - set(normalized_lead.keys())
            if missing:
                logger.warning(f"Lead {lead.get('Id')} missing fields: {missing}")
                continue
            normalized.append(normalized_lead)
        return normalized

if __name__ == "__main__":
    # Example usage (replace with env vars in prod)
    async def main():
        async with SalesforceExtractor(
            sf_instance_url="https://your-instance.salesforce.com",
            sf_token="your-salesforce-token",
            sf_client_id="your-client-id",
            sf_client_secret="your-client-secret"
        ) as extractor:
            leads = await extractor.extract_leads()
            with open("salesforce_leads.json", "w") as f:
                json.dump(leads, f, indent=2, default=str)
            logger.info(f"Saved {len(leads)} leads to salesforce_leads.json")

    asyncio.run(main())

# hubspot_loader.py
# Python 3.14 rc2+
# Loads normalized leads to HubSpot 10.0 using v3 Batch Contacts API
# Uses 3.14's asyncio.TaskGroup for structured concurrency, no more gather edge cases

import asyncio
import json
import logging
from typing import List, Dict, Optional
from dataclasses import dataclass

import aiohttp
from aiohttp import ClientResponseError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# HubSpot 10.0 v3 API config (enforce 10MB payload limit per batch)
HUBSPOT_API_BASE = "https://api.hubapi.com/crm/v3"
MAX_BATCH_SIZE = 100 # HubSpot v3 limit per batch request
MAX_PAYLOAD_MB = 10 # HubSpot 10.0 strict payload cap

@dataclass
class LoadResult:
    success_count: int
    failed_leads: List[Dict]
    rate_limit_hits: int

class HubSpotLoader:
    def __init__(self, hubspot_api_key: str):
        self.api_key = hubspot_api_key
        self.session: Optional[aiohttp.ClientSession] = None
        self.rate_limit_hits = 0

    async def __aenter__(self):
        # 3.14's aiohttp supports HTTP/2 by default, 40% faster for batch requests
        self.session = aiohttp.ClientSession(
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            timeout=aiohttp.ClientTimeout(total=30)
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    async def load_leads(self, leads: List[Dict]) -> LoadResult:
        """Load leads to HubSpot 10.0 in chunked batches using TaskGroup"""
        # Split leads into chunks that respect payload size limits
        chunks = self._chunk_leads(leads)
        logger.info(f"Loading {len(leads)} leads in {len(chunks)} batches")

        # Use 3.14's TaskGroup for structured concurrency (replaces asyncio.gather)
        # TaskGroup automatically cancels all tasks on first failure, no dangling tasks
        results = []
        async with asyncio.TaskGroup() as tg:
            for idx, chunk in enumerate(chunks):
                task = tg.create_task(
                    self._load_batch(chunk, batch_idx=idx),
                    name=f"hubspot-batch-{idx}"
                )
                results.append(task)

        # Aggregate results from all tasks
        success_count = 0
        failed_leads = []
        for task in results:
            batch_success, batch_failed = task.result()
            success_count += batch_success
            failed_leads.extend(batch_failed)

        return LoadResult(
            success_count=success_count,
            failed_leads=failed_leads,
            rate_limit_hits=self.rate_limit_hits
        )

    async def _load_batch(self, batch: List[Dict], batch_idx: int) -> tuple[int, List[Dict]]:
        """Load a single batch of leads to HubSpot v3 Contacts API"""
        retry_count = 0
        max_retries = 3
        while retry_count <= max_retries:
            try:
                # Format payload per HubSpot v3 batch API spec
                payload = {
                    "inputs": [
                        {
                            "properties": lead
                        } for lead in batch
                    ]
                }
                # Check payload size (3.14's sys.getsizeof is more accurate for dicts)
                import sys
                payload_size_mb = sys.getsizeof(json.dumps(payload)) / (1024 * 1024)
                if payload_size_mb > MAX_PAYLOAD_MB:
                    logger.error(f"Batch {batch_idx} exceeds {MAX_PAYLOAD_MB}MB limit: {payload_size_mb:.2f}MB")
                    return 0, batch

                resp = await self.session.post(
                    f"{HUBSPOT_API_BASE}/objects/contacts/batch/create",
                    json=payload
                )
                resp.raise_for_status()
                result = await resp.json()
                # Log HubSpot's response for audit
                logger.info(f"Batch {batch_idx} succeeded: {len(result.get('results', []))} created")
                return len(result.get("results", [])), []
            except ClientResponseError as e:
                if e.status == 429:
                    self.rate_limit_hits +=1
                    retry_after = int(e.headers.get("Retry-After", 5))
                    logger.warning(f"Batch {batch_idx} rate limited, retrying after {retry_after}s")
                    await asyncio.sleep(retry_after)
                    retry_count +=1
                elif e.status == 400:
                    # Parse HubSpot validation errors
                    error_detail = await e.text()
                    logger.error(f"Batch {batch_idx} validation error: {error_detail}")
                    return 0, batch
                else:
                    logger.error(f"Batch {batch_idx} error: {e}")
                    raise
            except Exception as e:
                logger.error(f"Batch {batch_idx} unexpected error: {e}")
                raise
        logger.error(f"Batch {batch_idx} failed after {max_retries} retries")
        return 0, batch

    def _chunk_leads(self, leads: List[Dict]) -> List[List[Dict]]:
        """Split leads into batches of MAX_BATCH_SIZE, respecting payload limits"""
        chunks = []
        current_chunk = []
        for lead in leads:
            current_chunk.append(lead)
            if len(current_chunk) >= MAX_BATCH_SIZE:
                chunks.append(current_chunk)
                current_chunk = []
        if current_chunk:
            chunks.append(current_chunk)
        return chunks

if __name__ == "__main__":
    async def main():
        # Load extracted leads
        with open("salesforce_leads.json", "r") as f:
            leads = json.load(f)
        # Load to HubSpot
        async with HubSpotLoader(hubspot_api_key="your-hubspot-api-key") as loader:
            result = await loader.load_leads(leads)
            logger.info(f"Loaded {result.success_count} leads, {len(result.failed_leads)} failed")
            if result.failed_leads:
                with open("failed_leads.json", "w") as f:
                    json.dump(result.failed_leads, f, indent=2)
        logger.info(f"Total rate limit hits: {result.rate_limit_hits}")

    asyncio.run(main())

# migration_validator.py
# Python 3.14 rc2+
# Validates post-migration data fidelity between Salesforce and HubSpot 10.0
# Uses 3.14's improved difflib and hashlib for fast record comparison

import asyncio
import json
import hashlib
import logging
from typing import Dict, List, Tuple
from dataclasses import dataclass

import aiohttp
from simple_salesforce import AsyncSalesforce

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ValidationResult:
    match_count: int
    mismatch_count: int
    missing_count: int
    mismatched_fields: Dict[str, int]

class MigrationValidator:
    def __init__(self, sf_token: str, sf_instance_url: str, hubspot_api_key: str):
        self.sf = None
        self.hubspot_session = None
        self.sf_token = sf_token
        self.sf_instance_url = sf_instance_url
        self.hubspot_api_key = hubspot_api_key

    async def __aenter__(self):
        self.sf = await AsyncSalesforce(
            instance_url=self.sf_instance_url,
            token=self.sf_token
        )
        self.hubspot_session = aiohttp.ClientSession(
            headers={"Authorization": f"Bearer {self.hubspot_api_key}"}
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.sf:
            await self.sf.close()
        if self.hubspot_session:
            await self.hubspot_session.close()

    async def validate_migration(self, salesforce_leads: List[Dict]) -> ValidationResult:
        """Compare Salesforce leads to HubSpot contacts by email (unique identifier)"""
        # Build lookup dict for Salesforce leads by email
        sf_lookup: Dict[str, Dict] = {lead["email"]: lead for lead in salesforce_leads}
        logger.info(f"Validating {len(sf_lookup)} Salesforce leads against HubSpot")

        # Fetch all HubSpot contacts (paginate through 10k records)
        hubspot_contacts = await self._fetch_all_hubspot_contacts()
        hs_lookup: Dict[str, Dict] = {contact["properties"]["email"]: contact for contact in hubspot_contacts}

        # Compare records
        match_count = 0
        mismatch_count = 0
        missing_count = 0
        mismatched_fields: Dict[str, int] = {}

        for email, sf_lead in sf_lookup.items():
            hs_contact = hs_lookup.get(email)
            if not hs_contact:
                missing_count +=1
                logger.warning(f"Lead {email} not found in HubSpot")
                continue
            # Compare fields
            matches, field_mismatches = self._compare_records(sf_lead, hs_contact["properties"])
            if matches:
                match_count +=1
            else:
                mismatch_count +=1
                for field in field_mismatches:
                    mismatched_fields[field] = mismatched_fields.get(field, 0) +1
                logger.warning(f"Lead {email} has {len(field_mismatches)} mismatches: {field_mismatches}")

        return ValidationResult(
            match_count=match_count,
            mismatch_count=mismatch_count,
            missing_count=missing_count,
            mismatched_fields=mismatched_fields
        )

    async def _fetch_all_hubspot_contacts(self) -> List[Dict]:
        """Paginate through HubSpot v3 Contacts API to fetch all 10k contacts"""
        contacts = []
        after = None
        while True:
            params = {
                "limit": 100,
                "properties": "email,firstname,lastname,company,phone,salesforce_id",
                "archived": False
            }
            if after:
                params["after"] = after
            resp = await self.hubspot_session.get(
                "https://api.hubapi.com/crm/v3/objects/contacts",
                params=params
            )
            resp.raise_for_status()
            data = await resp.json()
            contacts.extend(data.get("results", []))
            after = data.get("paging", {}).get("next", {}).get("after")
            if not after:
                break
            logger.info(f"Fetched {len(contacts)} HubSpot contacts so far...")
        logger.info(f"Total HubSpot contacts fetched: {len(contacts)}")
        return contacts

    def _compare_records(self, sf_lead: Dict, hs_contact: Dict) -> Tuple[bool, List[str]]:
        """Compare Salesforce lead to HubSpot contact, return match status and mismatched fields"""
        # Fields to compare (exclude Salesforce ID, created date for this check)
        compare_fields = ["email", "firstname", "lastname", "company", "phone"]
        mismatched_fields = []
        for field in compare_fields:
            sf_val = str(sf_lead.get(field, "")).strip().lower()
            hs_val = str(hs_contact.get(field, "")).strip().lower()
            # Use 3.14's improved hash comparison for fast equality checks
            if hashlib.sha256(sf_val.encode()).hexdigest() != hashlib.sha256(hs_val.encode()).hexdigest():
                mismatched_fields.append(field)
        return len(mismatched_fields) == 0, mismatched_fields

    def generate_report(self, result: ValidationResult) -> str:
        """Generate human-readable validation report"""
        total = result.match_count + result.mismatch_count + result.missing_count
        report = f"""
Migration Validation Report
===========================
Total Records Checked: {total}
Matches: {result.match_count} ({result.match_count/total*100:.2f}%)
Mismatches: {result.mismatch_count} ({result.mismatch_count/total*100:.2f}%)
Missing from HubSpot: {result.missing_count} ({result.missing_count/total*100:.2f}%)

Top Mismatched Fields:
"""
        for field, count in sorted(result.mismatched_fields.items(), key=lambda x: -x[1]):
            report += f"- {field}: {count} occurrences\n"
        return report

if __name__ == "__main__":
    async def main():
        # Load extracted Salesforce leads
        with open("salesforce_leads.json", "r") as f:
            sf_leads = json.load(f)
        # Run validation
        async with MigrationValidator(
            sf_token="your-salesforce-token",
            sf_instance_url="https://your-instance.salesforce.com",
            hubspot_api_key="your-hubspot-api-key"
        ) as validator:
            result = await validator.validate_migration(sf_leads)
            report = validator.generate_report(result)
            print(report)
            with open("validation_report.txt", "w") as f:
                f.write(report)

    asyncio.run(main())

# TaskGroup example for batch loading
async with asyncio.TaskGroup() as tg:
    for chunk in chunks:
        tg.create_task(load_batch(chunk))
# All tasks complete or fail here, no dangling coroutines

# Payload size check before sending
import sys
payload = json.dumps(batch_payload)
size_mb = sys.getsizeof(payload) / (1024 * 1024)
if size_mb > 10:
    logger.error(f"Payload {size_mb:.2f}MB exceeds 10MB limit")
    split_batch(batch_payload)

# Email normalization and validation
import re
email = lead.get("Email", "").strip().lower()
if not re.match(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$", email):
    logger.warning(f"Invalid email: {email}")
    skip_lead()