HubSpot API Rate Limiting: The 19/s Mistake

190 requests ÷ 10 seconds = 19 requests/second

import time

def make_request():
    # Naive implementation - DON'T DO THIS
    time.sleep(1/19)  # ~52ms between requests
    return api_call()

| 00:10 | 22 (spike) | 193 | ❌ 429 ERROR

import time
import threading
from collections import deque

class HubSpotRateLimiter:
    """
    Sliding window rate limiter for HubSpot API.

    Uses a token bucket approach that respects HubSpot's
    rolling 10-second window.
    """

    def __init__(self, max_requests: int = 190, window_seconds: int = 10):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = deque()  # Timestamps of recent requests
        self.lock = threading.Lock()

    def acquire(self) -> None:
        """
        Block until a request slot is available.
        Call this before every HubSpot API request.
        """
        while True:
            with self.lock:
                now = time.time()
                window_start = now - self.window_seconds

                # Remove timestamps outside the window
                while self.requests and self.requests[0] < window_start:
                    self.requests.popleft()

                # Check if we have capacity
                if len(self.requests) < self.max_requests:
                    self.requests.append(now)
                    return

                # Calculate wait time until oldest request exits window
                wait_time = self.requests[0] - window_start

            # Wait outside the lock
            time.sleep(wait_time + 0.01)  # Small buffer

    def get_remaining(self) -> int:
        """Return number of requests available in current window."""
        with self.lock:
            now = time.time()
            window_start = now - self.window_seconds

            while self.requests and self.requests[0] < window_start:
                self.requests.popleft()

            return self.max_requests - len(self.requests)


# Usage
rate_limiter = HubSpotRateLimiter(max_requests=190, window_seconds=10)

def get_contact(contact_id: str) -> dict:
    rate_limiter.acquire()  # Blocks until slot available
    return hubspot_client.crm.contacts.basic_api.get_by_id(contact_id)

import asyncio
import time
from collections import deque

class AsyncHubSpotRateLimiter:
    """Async-compatible sliding window rate limiter."""

    def __init__(self, max_requests: int = 190, window_seconds: int = 10):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = deque()
        self.lock = asyncio.Lock()

    async def acquire(self) -> None:
        """Async version - awaits until slot available."""
        while True:
            async with self.lock:
                now = time.time()
                window_start = now - self.window_seconds

                # Clean old timestamps
                while self.requests and self.requests[0] < window_start:
                    self.requests.popleft()

                if len(self.requests) < self.max_requests:
                    self.requests.append(now)
                    return

                wait_time = self.requests[0] - window_start

            await asyncio.sleep(wait_time + 0.01)


# Usage with aiohttp or httpx
limiter = AsyncHubSpotRateLimiter()

async def fetch_contacts(contact_ids: list[str]) -> list[dict]:
    async def fetch_one(contact_id: str) -> dict:
        await limiter.acquire()
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
                headers={"Authorization": f"Bearer {access_token}"}
            )
            return response.json()

    return await asyncio.gather(*[fetch_one(cid) for cid in contact_ids])

class HubSpotRateLimiter {
  constructor(maxRequests = 190, windowSeconds = 10) {
    this.maxRequests = maxRequests;
    this.windowMs = windowSeconds * 1000;
    this.requests = [];
  }

  async acquire() {
    while (true) {
      const now = Date.now();
      const windowStart = now - this.windowMs;

      // Remove old timestamps
      this.requests = this.requests.filter(ts => ts > windowStart);

      if (this.requests.length < this.maxRequests) {
        this.requests.push(now);
        return;
      }

      // Wait for oldest request to exit window
      const waitTime = this.requests[0] - windowStart + 10;
      await new Promise(resolve => setTimeout(resolve, waitTime));
    }
  }
}

// Usage
const limiter = new HubSpotRateLimiter();

async function getContact(contactId) {
  await limiter.acquire();
  const response = await hubspotClient.crm.contacts.basicApi.getById(contactId);
  return response;
}

def make_request_adaptive(url: str, rate_limiter: HubSpotRateLimiter) -> dict:
    rate_limiter.acquire()
    response = requests.get(url, headers=auth_headers)

    # Check remaining capacity
    remaining = int(response.headers.get('X-HubSpot-RateLimit-Interval-Remaining', 0))

    if remaining < 20:
        # Getting close to limit - slow down
        time.sleep(0.5)

    if response.status_code == 429:
        # Hit the limit - back off
        retry_after = int(response.headers.get('Retry-After', 10))
        time.sleep(retry_after)
        return make_request_adaptive(url, rate_limiter)  # Retry

    return response.json()

# BAD: 100 API calls
for contact_id in contact_ids[:100]:
    contact = client.crm.contacts.basic_api.get_by_id(contact_id)

# GOOD: 1 API call
from hubspot.crm.contacts import BatchReadInputSimplePublicObjectId

batch_input = BatchReadInputSimplePublicObjectId(
    inputs=[{"id": cid} for cid in contact_ids[:100]],
    properties=["email", "firstname", "lastname"]
)
results = client.crm.contacts.batch_api.read(batch_input)

from hubspot.crm.contacts import BatchInputSimplePublicObjectInputForCreate

batch_input = BatchInputSimplePublicObjectInputForCreate(
    inputs=[
        {"properties": {"email": "[email protected]", "firstname": "Alice"}},
        {"properties": {"email": "[email protected]", "firstname": "Bob"}},
        # ... up to 100
    ]
)
results = client.crm.contacts.batch_api.create(batch_input)

# Get all contacts updated in last 24 hours
from datetime import datetime, timedelta

yesterday = int((datetime.now() - timedelta(days=1)).timestamp() * 1000)

search_request = {
    "filterGroups": [{
        "filters": [{
            "propertyName": "lastmodifieddate",
            "operator": "GTE",
            "value": str(yesterday)
        }]
    }],
    "limit": 200,
    "properties": ["email", "firstname", "lastname"]
}

results = client.crm.contacts.search_api.do_search(search_request)

import time
from functools import wraps

def retry_on_rate_limit(max_retries: int = 3):
    """Decorator to handle HubSpot 429 errors with exponential backoff."""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except ApiException as e:
                    if e.status == 429:
                        # Get retry-after from headers, default to exponential backoff
                        retry_after = int(e.headers.get('Retry-After', 2 ** attempt))
                        print(f"Rate limited. Waiting {retry_after}s (attempt {attempt + 1})")
                        time.sleep(retry_after)
                    else:
                        raise
            raise Exception(f"Max retries ({max_retries}) exceeded")
        return wrapper
    return decorator


@retry_on_rate_limit(max_retries=5)
def get_contact_safe(contact_id: str) -> dict:
    rate_limiter.acquire()
    return client.crm.contacts.basic_api.get_by_id(contact_id)

import redis

class DistributedRateLimiter:
    """Redis-backed rate limiter for multi-service environments."""

    def __init__(self, redis_client, key: str, max_requests: int = 190, window: int = 10):
        self.redis = redis_client
        self.key = f"hubspot_ratelimit:{key}"
        self.max_requests = max_requests
        self.window = window

    def acquire(self) -> bool:
        """
        Attempt to acquire a slot. Returns True if successful.
        Uses Redis sorted sets for distributed sliding window.
        """
        now = time.time()
        window_start = now - self.window

        pipe = self.redis.pipeline()

        # Remove old entries
        pipe.zremrangebyscore(self.key, 0, window_start)

        # Count current entries
        pipe.zcard(self.key)

        # Add new entry (optimistically)
        pipe.zadd(self.key, {str(now): now})

        # Set expiry on the key
        pipe.expire(self.key, self.window + 1)

        results = pipe.execute()
        current_count = results[1]

        if current_count >= self.max_requests:
            # Remove the entry we just added
            self.redis.zrem(self.key, str(now))
            return False

        return True

    def acquire_blocking(self, timeout: float = 30) -> bool:
        """Block until slot available or timeout."""
        start = time.time()
        while time.time() - start < timeout:
            if self.acquire():
                return True
            time.sleep(0.1)
        return False

import asyncio
import httpx
from collections import deque
import time

class AsyncHubSpotRateLimiter:
    def __init__(self, max_requests: int = 170, window_seconds: int = 10):
        # Using 170 instead of 190 for safety margin
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = deque()
        self.lock = asyncio.Lock()

    async def acquire(self):
        while True:
            async with self.lock:
                now = time.time()
                window_start = now - self.window_seconds

                while self.requests and self.requests[0] < window_start:
                    self.requests.popleft()

                if len(self.requests) < self.max_requests:
                    self.requests.append(now)
                    return

                wait_time = self.requests[0] - window_start

            await asyncio.sleep(wait_time + 0.01)


async def audit_consent_preferences(contact_ids: list[str]) -> dict:
    """
    Audit subscription preferences for all contacts.

    With 400,000 contacts at ~17 requests/second,
    this completes in approximately 6.5 hours.
    """
    limiter = AsyncHubSpotRateLimiter(max_requests=170)
    results = {"subscribed": 0, "unsubscribed": 0, "errors": 0}

    async with httpx.AsyncClient() as client:

        async def check_one(contact_id: str) -> str:
            await limiter.acquire()
            try:
                response = await client.get(
                    f"https://api.hubapi.com/communication-preferences/v3/status/email/{contact_id}",
                    headers={"Authorization": f"Bearer {ACCESS_TOKEN}"},
                    timeout=30.0
                )
                if response.status_code == 200:
                    data = response.json()
                    # Check marketing subscription (ID varies by portal)
                    for sub in data.get("subscriptionStatuses", []):
                        if sub["id"] == MARKETING_SUBSCRIPTION_ID:
                            return "subscribed" if sub["status"] == "SUBSCRIBED" else "unsubscribed"
                return "unknown"
            except Exception as e:
                print(f"Error for {contact_id}: {e}")
                return "error"

        # Process in chunks to avoid memory issues
        chunk_size = 1000
        for i in range(0, len(contact_ids), chunk_size):
            chunk = contact_ids[i:i + chunk_size]
            chunk_results = await asyncio.gather(*[check_one(cid) for cid in chunk])

            for result in chunk_results:
                if result in results:
                    results[result] += 1
                else:
                    results["errors"] += 1

            # Progress update
            processed = min(i + chunk_size, len(contact_ids))
            print(f"Processed {processed:,}/{len(contact_ids):,} contacts")

    return results


# Run the audit
if __name__ == "__main__":
    # Load contact IDs from your source
    contact_ids = load_contact_ids()  # Your implementation

    results = asyncio.run(audit_consent_preferences(contact_ids))
    print(f"Audit complete: {results}")