hubspot-warehouse-sync

What this skill does

# HubSpot Warehouse Sync

## Overview

Move HubSpot CRM data to BigQuery, Snowflake, or Postgres in a way that survives production — not just the demo. This is not a connector walkthrough. It is the extraction and load code your pipeline runs when a 2M-contact backfill burns through the 500K daily call quota at noon, when CDC misses three days of deal updates because association changes do not update `hs_lastmodifieddate`, when a portal admin adds a custom property and your warehouse table schema silently drifts, and when a network timeout at record 45,000 causes your retry to insert 100 duplicate rows.

The six production failures this skill prevents:

1. **Backfill exhausting the daily rate limit before completing** — 2M contacts at 100 records/call is 20,000 calls. At 100 calls/10s the math says 33 minutes, but that burns your entire 500K daily quota before noon and takes every other integration down with it. Token bucket rate limiting with a configurable daily ceiling is non-optional.
2. **CDC missing association changes** — HubSpot's `hs_lastmodifieddate` is updated when any property on the contact record changes, but not when an association is created or deleted. A contact-to-deal link added by a sales rep is invisible to a lastmodifieddate-based incremental poll. Association CDC requires a separate poll strategy.
3. **Schema drift causing silent extraction failures** — when a portal admin adds or removes a custom property, the warehouse table schema and the extraction property list become misaligned. New properties are dropped on the floor. Removed properties cause KeyError on row construction. Neither failure raises an alarm without explicit schema validation.
4. **Duplicate rows on batch retry** — a network failure mid-batch causes the batch to be re-sent. Without a proper upsert key the warehouse gets duplicate rows that are invisible until an analyst notices double-counted revenue. The correct upsert key for contacts is `id` (HubSpot object ID), not a composite of name/email.
5. **Large payload failures on associated object inline pulls** — contacts with thousands of engagement records cause payload sizes that exceed HTTP response limits when associations are pulled inline. Associations must be fetched in a separate batch read pass.
6. **Timezone inconsistency in aggregations** — HubSpot stores all timestamps as Unix milliseconds in UTC. If the warehouse session timezone is set to a local timezone, `DATE(created_at)` aggregations produce different daily totals depending on where the analyst runs the query.

## Prerequisites

- Python 3.10+
- HubSpot private app token with scopes: `crm.objects.contacts.read`, `crm.objects.companies.read`, `crm.objects.deals.read`, `crm.associations.read`, `crm.schemas.contacts.read`
- Target warehouse credentials:
  - BigQuery: service account JSON with `bigquery.dataEditor` role
  - Snowflake: user/password or key-pair auth, warehouse + database + schema
  - Postgres: connection string with write access to the target schema
- Python packages: `requests`, `pandas`, `pyarrow`, `google-cloud-bigquery` (BigQuery) or `snowflake-connector-python` (Snowflake) or `psycopg2-binary` (Postgres)
- A secret store the runtime can read (env vars, GCP Secret Manager, AWS Secrets Manager)

## Instructions

Build in this order. Each section neutralizes one production failure mode.

### 1. Token bucket rate limiter (neutralizes backfill quota burnout)

Naive extraction loops call the API as fast as possible. At 100 calls/10s with 20K calls needed, you burn 200K of your 500K daily quota in 33 minutes — and that is before any CDC polling or webhook processing. A production backfill must budget its calls over the full day so other integrations still have headroom.

The token bucket strategy: set a `daily_budget` ceiling below the account limit (e.g., 400K of 500K), compute how many calls per second that allows over 24 hours, and enforce a minimum interval between calls. Burn-rate is checked against live rate-limit headers on every response.

```python
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()
```

### 2. Full backfill with cursor pagination (contacts search API)

The search API (`POST /crm/v3/objects/contacts/search`) supports cursor pagination via `after`. It returns a maximum of 100 records per page and up to 10,000 records total per search. For tables larger than 10K records, use the `lastmodifieddate` range-slicing strategy: paginate within 30-day windows, sliding forward from the earliest `hs_createdate` in the portal.

```python
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("res