Hubspot Warehouse Sync

import time
import threading
from dataclasses import dataclass, field

@dataclass
class TokenBucket:
    """
    Token bucket rate limiter for HubSpot API calls.

    daily_budget: max calls to spend per 24h window (stay below 500K limit)
    burst_limit: max calls per 10s window (100 for most tiers)
    """
    daily_budget: int = 400_000
    burst_limit: int = 95          # leave 5 calls headroom per 10s window
    _lock: threading.Lock = field(default_factory=threading.Lock)
    _calls_today: int = 0
    _window_calls: int = 0
    _window_start: float = field(default_factory=time.monotonic)
    _day_start: float = field(default_factory=time.monotonic)

    def acquire(self) -> None:
        with self._lock:
            now = time.monotonic()

            # Reset daily counter
            if now - self._day_start >= 86_400:
                self._calls_today = 0
                self._day_start = now

            # Reset 10s window counter
            if now - self._window_start >= 10.0:
                self._window_calls = 0
                self._window_start = now

            # Hard stop if daily budget exhausted — do not burn other integrations
            if self._calls_today >= self.daily_budget:
                seconds_left = 86_400 - (now - self._day_start)
                raise DailyBudgetExhausted(
                    f"Daily call budget of {self.daily_budget:,} reached. "
                    f"Resuming in {seconds_left/3600:.1f}h."
                )

            # Burst window throttle — sleep until window resets if full
            if self._window_calls >= self.burst_limit:
                sleep_s = 10.0 - (now - self._window_start) + 0.1
                time.sleep(max(0, sleep_s))
                self._window_calls = 0
                self._window_start = time.monotonic()

            self._calls_today += 1
            self._window_calls += 1

    def update_from_headers(self, headers: dict) -> None:
        """Adjust pacing from live rate-limit headers on each response."""
        remaining = int(headers.get("X-HubSpot-RateLimit-Daily-Remaining", self.daily_budget))
        if remaining < 50_000:
            # Emergency throttle: burn rate is too high, cut burst limit in half
            with self._lock:
                self.burst_limit = max(10, self.burst_limit // 2)

class DailyBudgetExhausted(Exception):
    pass

RATE_LIMITER = TokenBucket()

import requests
import json

BASE_URL = "https://api.hubapi.com"

def fetch_contacts_page(
    token: str,
    after: str | None,
    properties: list[str],
    rate_limiter: TokenBucket,
) -> tuple[list[dict], str | None]:
    """Fetch one page of contacts. Returns (records, next_cursor)."""
    RATE_LIMITER.acquire()

    body = {
        "limit": 100,
        "properties": properties,
        "sorts": [{"propertyName": "hs_lastmodifieddate", "direction": "ASCENDING"}],
    }
    if after:
        body["after"] = after

    resp = requests.post(
        f"{BASE_URL}/crm/v3/objects/contacts/search",
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        json=body,
        timeout=30,
    )
    rate_limiter.update_from_headers(dict(resp.headers))

    if resp.status_code == 429:
        retry_after = int(resp.headers.get("Retry-After", 10))
        time.sleep(retry_after)
        return fetch_contacts_page(token, after, properties, rate_limiter)  # one retry

    resp.raise_for_status()
    data = resp.json()

    results = data.get("results", [])
    next_cursor = data.get("paging", {}).get("next", {}).get("after")
    return results, next_cursor


def backfill_contacts(
    token: str,
    properties: list[str],
    rate_limiter: TokenBucket,
    checkpoint_file: str = "/tmp/hubspot_backfill_checkpoint.json",
) -> int:
    """
    Full backfill with checkpoint. Resume-safe: reads cursor from checkpoint_file
    so a mid-run failure restarts from the last completed page, not from zero.

    Returns total records written.
    """
    # Load checkpoint
    try:
        with open(checkpoint_file) as f:
            checkpoint = json.load(f)
        after = checkpoint.get("after")
        total = checkpoint.get("total", 0)
        print(f"Resuming backfill from cursor {after}, {total:,} records already written")
    except FileNotFoundError:
        after = None
        total = 0

    while True:
        records, next_cursor = fetch_contacts_page(token, after, properties, rate_limiter)

        if not records:
            break

        yield records   # caller handles warehouse write

        total += len(records)
        after = next_cursor

        # Write checkpoint after every page so a restart costs at most 100 records
        with open(checkpoint_file, "w") as f:
            json.dump({"after": after, "total": total}, f)

        if not next_cursor:
            break

    print(f"Backfill complete: {total:,} contacts")
    return total

import datetime

def incremental_contacts_since(
    token: str,
    since_ms: int,
    properties: list[str],
    rate_limiter: TokenBucket,
) -> list[dict]:
    """
    CDC property poll: return all contacts modified after since_ms (Unix ms UTC).

    Does NOT cover association changes — run poll_association_changes() separately.
    """
    all_records = []
    after = None

    while True:
        RATE_LIMITER.acquire()
        body = {
            "limit": 100,
            "properties": properties,
            "filterGroups": [{
                "filters": [{
                    "propertyName": "hs_lastmodifieddate",
                    "operator": "GT",
                    "value": str(since_ms),
                }]
            }],
            "sorts": [{"propertyName": "hs_lastmodifieddate", "direction": "ASCENDING"}],
        }
        if after:
            body["after"] = after

        resp = requests.post(
            f"{BASE_URL}/crm/v3/objects/contacts/search",
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
            json=body,
            timeout=30,
        )
        rate_limiter.update_from_headers(dict(resp.headers))
        resp.raise_for_status()

        data = resp.json()
        all_records.extend(data.get("results", []))
        next_cursor = data.get("paging", {}).get("next", {}).get("after")
        if not next_cursor:
            break
        after = next_cursor

    return all_records


def poll_association_changes(
    token: str,
    contact_ids: list[str],
    rate_limiter: TokenBucket,
) -> dict[str, list[str]]:
    """
    Fetch current contact → deal associations for a list of contact IDs.
    Use this to detect additions and deletions that hs_lastmodifieddate misses.

    Strategy: compare fetched associations against warehouse snapshot.
    Differences = changes that must be written.

    Returns {contact_id: [deal_id, ...]}
    """
    # Batch read associations: up to 100 contacts per call (v4 associations API)
    result = {}
    for chunk_start in range(0, len(contact_ids), 100):
        chunk = contact_ids[chunk_start:chunk_start + 100]
        RATE_LIMITER.acquire()

        resp = requests.post(
            f"{BASE_URL}/crm/v4/associations/contacts/deals/batch/read",
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
            json={"inputs": [{"id": cid} for cid in chunk]},
            timeout=30,
        )
        rate_limiter.update_from_headers(dict(resp.headers))
        resp.raise_for_status()

        for item in resp.json().get("results", []):
            from_id = str(item["from"]["id"])
            to_ids = [str(a["toObjectId"]) for a in item.get("to", [])]
            result[from_id] = to_ids

    return result

def fetch_hubspot_property_schema(token: str) -> dict[str, str]:
    """
    Returns {property_name: warehouse_type} for all contact properties.
    Maps HubSpot field types to warehouse-appropriate column types.
    """
    resp = requests.get(
        f"{BASE_URL}/crm/v3/properties/contacts",
        headers={"Authorization": f"Bearer {token}"},
        timeout=30,
    )
    resp.raise_for_status()

    TYPE_MAP = {
        "string":      "TEXT",
        "number":      "FLOAT64",
        "date":        "DATE",
        "datetime":    "TIMESTAMP",
        "bool":        "BOOLEAN",
        "enumeration": "TEXT",
        "phone_number":"TEXT",
        "json":        "TEXT",       # store complex types as serialized JSON
    }

    schema = {}
    for prop in resp.json().get("results", []):
        hs_type = prop.get("type", "string")
        schema[prop["name"]] = TYPE_MAP.get(hs_type, "TEXT")

    return schema


def detect_schema_drift(
    hubspot_schema: dict[str, str],
    warehouse_columns: set[str],
) -> tuple[dict[str, str], set[str]]:
    """
    Returns:
      added:   {column_name: type}  — in HubSpot but not warehouse
      removed: {column_name}        — in warehouse but not HubSpot
    """
    hs_columns = set(hubspot_schema.keys())
    added   = {k: v for k, v in hubspot_schema.items() if k not in warehouse_columns}
    removed = warehouse_columns - hs_columns - {"_synced_at", "_sync_run_id"}  # exclude meta cols
    return added, removed


def generate_alter_statements(table: str, added: dict[str, str]) -> list[str]:
    """Generate ALTER TABLE ADD COLUMN statements for new HubSpot properties."""
    stmts = []
    for col, dtype in added.items():
        safe_col = col.replace("-", "_").lower()
        stmts.append(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS {safe_col} {dtype};")
    return stmts