""" HubSpot Sequences → Supabase (incremental since a millisecond cursor) Sequences are fetched per-user via the HubSpot automation REST API (no SDK search endpoint available). The cursor filters out sequences whose updatedAt is at or before since_ms, so only new or changed sequences are upserted each run. Usage from orchestrator: import hubspot_sequences hubspot_sequences.main(since_ms=) Direct CLI: # epoch ms python hubspot_sequences.py 1754025600000 # ISO-8601 python hubspot_sequences.py 2026-01-01T00:00:00Z # Back-compat date (floors to 00:00Z) python hubspot_sequences.py 2026-01-01 """ import os import re import time import logging import datetime from typing import Dict, List, Optional, Tuple, Union import httpx from dotenv import load_dotenv from supabase import create_client from hubspot_utils import parse_ts, try_parse_int from supabase_utils import insert_into_supabase_table, update_sync_metadata # ----------------------------------------------------------------------------- # Constants # ----------------------------------------------------------------------------- HUBSPOT_BASE_URL = "https://api.hubapi.com" HUBSPOT_TIMEOUT = 30.0 HUBSPOT_PAGE_LIMIT = 100 REQUEST_SLEEP_SECONDS = 0.1 MAX_RETRIES = 3 # ----------------------------------------------------------------------------- # Logging # ----------------------------------------------------------------------------- logging.basicConfig( filename=( f"logs/hubspot_sequences_pipeline_" f"{datetime.datetime.now().strftime('%Y-%m-%d')}.log" ), filemode="a", level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) # ----------------------------------------------------------------------------- # Environment # ----------------------------------------------------------------------------- load_dotenv() HUBSPOT_TOKEN = os.getenv("HUBSPOT_TOKEN") SUPABASE_URL = os.getenv("SUPABASE_URL") SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY") # Optional bootstrap cursor if orchestrator doesn't provide one BOOTSTRAP_SINCE_MS_ENV = os.getenv("HUBSPOT_SEQUENCES_SINCE_MS") if not HUBSPOT_TOKEN: raise RuntimeError("HUBSPOT_TOKEN is not set") if not SUPABASE_URL or not SUPABASE_SERVICE_ROLE_KEY: raise RuntimeError("Supabase env vars are not set") supabase_client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY) # ----------------------------------------------------------------------------- # Time helpers # ----------------------------------------------------------------------------- def _ensure_utc(dt: datetime.datetime) -> datetime.datetime: if dt.tzinfo is None: dt = dt.replace(tzinfo=datetime.timezone.utc) return dt.astimezone(datetime.timezone.utc) def floor_to_utc_midnight(dt: datetime.datetime) -> datetime.datetime: dt = _ensure_utc(dt) return dt.replace(hour=0, minute=0, second=0, microsecond=0) def _parse_iso_like_to_dt(value: str) -> datetime.datetime: if isinstance(value, str) and value.endswith("Z"): value = value[:-1] + "+00:00" dt = datetime.datetime.fromisoformat(value) return _ensure_utc(dt) def to_epoch_ms(dt_or_str: Union[str, datetime.datetime]) -> int: if isinstance(dt_or_str, str): dt = _parse_iso_like_to_dt(dt_or_str) elif isinstance(dt_or_str, datetime.datetime): dt = _ensure_utc(dt_or_str) else: raise TypeError(f"Unsupported type for to_epoch_ms: {type(dt_or_str)}") return int(dt.timestamp() * 1000) def parse_any_ts_ms(value: Optional[Union[str, int, float]]) -> Optional[int]: """ Accepts ms-epoch / sec-epoch / ISO-8601; returns ms since epoch or None. """ if value is None: return None try: v = int(str(value)) if v < 10_000_000_000_000: # seconds → ms v *= 1000 return v except ValueError: pass try: return to_epoch_ms(str(value)) except Exception: logging.warning("Could not parse timestamp value=%r", value) return None # ----------------------------------------------------------------------------- # HubSpot HTTP helpers # ----------------------------------------------------------------------------- def _get_hubspot_headers() -> Dict[str, str]: return { "Authorization": f"Bearer {HUBSPOT_TOKEN}", "Content-Type": "application/json", } def hubspot_get( client: httpx.Client, path: str, params: Optional[Dict] = None, max_retries: int = MAX_RETRIES, ) -> Optional[Dict]: """ Make a GET request to the HubSpot API with retry and rate-limit handling. :param client: The HTTPX client to use. :param path: API path (appended to HUBSPOT_BASE_URL). :param params: Optional query parameters. :param max_retries: Maximum number of retry attempts. :return: Parsed JSON response dict, or None if all retries fail. """ url = f"{HUBSPOT_BASE_URL}{path}" for attempt in range(1, max_retries + 1): try: response = client.get(url, headers=_get_hubspot_headers(), params=params) if response.status_code == 429: retry_after = response.headers.get("Retry-After") sleep_seconds = float(retry_after) if retry_after else float(attempt) logging.warning( "Rate limited on %s. Sleeping for %s seconds.", url, sleep_seconds ) time.sleep(sleep_seconds) continue response.raise_for_status() return response.json() except httpx.HTTPStatusError as exc: logging.error( "HubSpot request failed [%s] %s: %s", exc.response.status_code, url, exc.response.text, ) if attempt == max_retries: raise time.sleep(attempt) except httpx.RequestError as exc: logging.error("HubSpot request error on %s: %s", url, exc) if attempt == max_retries: raise time.sleep(attempt) return None # ----------------------------------------------------------------------------- # Supabase user lookup # ----------------------------------------------------------------------------- def fetch_hubspot_user_ids() -> List[int]: """ Fetch distinct, non-null user IDs from the hubspot_users table in Supabase. :return: List of integer user IDs. """ response = ( supabase_client.table("hubspot_users") .select("user_id") .not_.is_("user_id", "null") .execute() ) rows = response.data or [] user_ids: List[int] = [] seen: set = set() for row in rows: user_id = try_parse_int(row.get("user_id")) if user_id is None or user_id in seen: continue seen.add(user_id) user_ids.append(user_id) return user_ids # ----------------------------------------------------------------------------- # HubSpot sequence fetchers # ----------------------------------------------------------------------------- def fetch_sequence_summaries_for_user( client: httpx.Client, user_id: int, ) -> List[Dict]: """ Page through all sequence summaries for a given HubSpot user. :param client: The HTTPX client to use. :param user_id: HubSpot user ID to scope the request. :return: List of raw sequence summary dicts. """ summaries: List[Dict] = [] after: Optional[str] = None while True: params: Dict = {"limit": HUBSPOT_PAGE_LIMIT, "userId": user_id} if after is not None: params["after"] = after payload = hubspot_get(client, "/automation/v4/sequences", params=params) results = payload.get("results", []) if payload else [] summaries.extend(results) paging = payload.get("paging", {}) if payload else {} after = (paging.get("next") or {}).get("after") if not after: break time.sleep(REQUEST_SLEEP_SECONDS) return summaries def fetch_sequence_detail( client: httpx.Client, sequence_id: int, user_id: int, ) -> Optional[Dict]: """ Fetch the full detail payload for a single sequence. :param client: The HTTPX client to use. :param sequence_id: The sequence ID to fetch. :param user_id: HubSpot user ID to scope the request. :return: Raw sequence detail dict, or None if the request fails. """ return hubspot_get( client, f"/automation/v4/sequences/{sequence_id}", params={"userId": user_id}, ) # ----------------------------------------------------------------------------- # Transform helpers # ----------------------------------------------------------------------------- def transform_sequence(sequence_detail: Dict) -> Dict: """ Flatten a HubSpot sequence detail dict into a Supabase row dict. :param sequence_detail: Raw sequence detail from the HubSpot API. :return: Flattened sequence record ready for upsert. """ settings = sequence_detail.get("settings") or {} return { "sequence_id": try_parse_int(sequence_detail.get("id")), "name": sequence_detail.get("name"), "user_id": try_parse_int(sequence_detail.get("userId")), "hubspot_created_at": parse_ts(sequence_detail.get("createdAt")), "hubspot_updated_at": parse_ts(sequence_detail.get("updatedAt")), "settings_id": try_parse_int(settings.get("id")), "eligible_follow_up_days": settings.get("eligibleFollowUpDays"), "selling_strategy": settings.get("sellingStrategy"), "send_window_start_minute": try_parse_int(settings.get("sendWindowStartMinute")), "send_window_end_minute": try_parse_int(settings.get("sendWindowEndMinute")), "task_reminder_minute": try_parse_int(settings.get("taskReminderMinute")), "individual_task_reminders_enabled": settings.get("individualTaskRemindersEnabled"), "settings_hubspot_created_at": parse_ts(settings.get("createdAt")), "settings_hubspot_updated_at": parse_ts(settings.get("updatedAt")), } def transform_steps(sequence_detail: Dict) -> List[Dict]: """ Flatten all steps from a sequence detail into a list of Supabase row dicts. :param sequence_detail: Raw sequence detail from the HubSpot API. :return: List of flattened step records ready for upsert. """ sequence_id = try_parse_int(sequence_detail.get("id")) steps: List[Dict] = [] for step in sequence_detail.get("steps", []): task_pattern = step.get("taskPattern") or {} email_pattern = step.get("emailPattern") or {} steps.append({ "step_id": try_parse_int(step.get("id")), "sequence_id": sequence_id, "step_order": try_parse_int(step.get("stepOrder")), "delay_millis": try_parse_int(step.get("delayMillis")), "action_type": step.get("actionType"), "hubspot_created_at": parse_ts(step.get("createdAt")), "hubspot_updated_at": parse_ts(step.get("updatedAt")), "task_pattern_id": try_parse_int(task_pattern.get("id")), "task_type": task_pattern.get("taskType"), "task_priority": task_pattern.get("taskPriority"), "subject": task_pattern.get("subject"), "notes": task_pattern.get("notes"), "task_pattern_hubspot_created_at": parse_ts(task_pattern.get("createdAt")), "task_pattern_hubspot_updated_at": parse_ts(task_pattern.get("updatedAt")), "email_pattern_id": try_parse_int(email_pattern.get("id")), "email_template_id": try_parse_int(email_pattern.get("templateId")), "email_pattern_hubspot_created_at": parse_ts(email_pattern.get("createdAt")), "email_pattern_hubspot_updated_at": parse_ts(email_pattern.get("updatedAt")), "thread_email_to_step_order": try_parse_int( email_pattern.get("threadEmailToStepOrder") ), }) return steps def transform_dependencies(sequence_detail: Dict) -> List[Dict]: """ Flatten all dependencies from a sequence detail into a list of Supabase row dicts. :param sequence_detail: Raw sequence detail from the HubSpot API. :return: List of flattened dependency records ready for upsert. """ sequence_id = try_parse_int(sequence_detail.get("id")) dependencies: List[Dict] = [] for dependency in sequence_detail.get("dependencies", []): dependencies.append({ "dependency_id": try_parse_int(dependency.get("id")), "sequence_id": sequence_id, "dependency_type": dependency.get("dependencyType"), "required_by_sequence_step_id": try_parse_int( dependency.get("requiredBySequenceStepId") ), "relies_on_sequence_step_id": try_parse_int( dependency.get("reliesOnSequenceStepId") ), "required_by_step_order": try_parse_int(dependency.get("requiredByStepOrder")), "relies_on_step_order": try_parse_int(dependency.get("reliesOnStepOrder")), "hubspot_created_at": parse_ts(dependency.get("createdAt")), "hubspot_updated_at": parse_ts(dependency.get("updatedAt")), }) return dependencies # ----------------------------------------------------------------------------- # Fetch + transform all sequences (with cursor filtering) # ----------------------------------------------------------------------------- def fetch_all_sequence_details( since_ms: int, ) -> Tuple[List[Dict], List[Dict], List[Dict], Optional[int]]: """ Fetch all sequence details from HubSpot for every user in Supabase, filtering to sequences updated after since_ms. The HubSpot sequences API does not support server-side timestamp filtering, so summaries are fetched per user and filtered client-side by updatedAt. :param since_ms: Only include sequences whose updatedAt > since_ms (epoch ms). :return: (sequences, steps, dependencies, max_updated_at_ms) where max_updated_at_ms is the highest updatedAt seen, or None. """ user_ids = fetch_hubspot_user_ids() logging.info("Fetched %d HubSpot user IDs from Supabase.", len(user_ids)) if not user_ids: return [], [], [], None sequences: List[Dict] = [] steps: List[Dict] = [] dependencies: List[Dict] = [] processed_sequence_ids: set = set() max_updated_at_ms: Optional[int] = None with httpx.Client(timeout=HUBSPOT_TIMEOUT) as client: for user_id in user_ids: logging.info("Fetching sequence summaries for user_id=%s", user_id) summaries = fetch_sequence_summaries_for_user(client, user_id) logging.info( "Fetched %d sequence summaries for user_id=%s", len(summaries), user_id ) for summary in summaries: sequence_id = try_parse_int(summary.get("id")) if sequence_id is None or sequence_id in processed_sequence_ids: continue # Client-side cursor filter: skip sequences not updated since cursor updated_at_ms = parse_any_ts_ms(summary.get("updatedAt")) if updated_at_ms is not None and updated_at_ms <= since_ms: continue logging.info("Fetching sequence detail for sequence_id=%s", sequence_id) detail = fetch_sequence_detail(client, sequence_id, user_id) if not detail: logging.warning( "No detail payload returned for sequence_id=%s", sequence_id ) continue processed_sequence_ids.add(sequence_id) # Track max updatedAt for cursor advancement if updated_at_ms is not None and ( max_updated_at_ms is None or updated_at_ms > max_updated_at_ms ): max_updated_at_ms = updated_at_ms sequences.append(transform_sequence(detail)) steps.extend(transform_steps(detail)) dependencies.extend(transform_dependencies(detail)) time.sleep(REQUEST_SLEEP_SECONDS) return sequences, steps, dependencies, max_updated_at_ms # ----------------------------------------------------------------------------- # Upsert # ----------------------------------------------------------------------------- def upsert_sequences( sequences: List[Dict], steps: List[Dict], dependencies: List[Dict], ) -> None: if sequences: insert_into_supabase_table(supabase_client, "hubspot_sequences", sequences) print(f"Upserted {len(sequences)} sequences.") if steps: insert_into_supabase_table(supabase_client, "hubspot_sequence_steps", steps) print(f"Upserted {len(steps)} sequence steps.") if dependencies: insert_into_supabase_table( supabase_client, "hubspot_sequence_dependencies", dependencies ) print(f"Upserted {len(dependencies)} sequence dependencies.") # ----------------------------------------------------------------------------- # Main (timestamp cursor) # ----------------------------------------------------------------------------- def main(since_ms: Optional[int] = None) -> None: """ Orchestrates: 1) Fetch sequence details per user, filtered by since_ms cursor 2) Upsert sequences, steps, and dependencies into Supabase 3) Update sync metadata for all three tables """ # Resolve since_ms if since_ms is None and BOOTSTRAP_SINCE_MS_ENV: try: since_ms = int(BOOTSTRAP_SINCE_MS_ENV) except ValueError: raise RuntimeError("HUBSPOT_SEQUENCES_SINCE_MS must be an integer (ms) if set.") if since_ms is None: # Default: today@00:00:00Z for first run today0 = floor_to_utc_midnight(datetime.datetime.now(datetime.timezone.utc)) since_ms = to_epoch_ms(today0) print(f"Fetching sequences updated after {since_ms} ms ...") logging.info("Fetching sequences updated after %d ms.", since_ms) sequences, steps, dependencies, max_updated_at_ms = fetch_all_sequence_details(since_ms) now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat() if not sequences: print("No sequences found beyond the cursor. Updating sync metadata and exiting.") logging.warning("No sequences returned beyond cursor %d.", since_ms) update_sync_metadata(supabase_client, "hubspot_sequences", now_iso) update_sync_metadata(supabase_client, "hubspot_sequence_steps", now_iso) update_sync_metadata(supabase_client, "hubspot_sequence_dependencies", now_iso) return print(f"Fetched {len(sequences)} sequences from HubSpot.") print(f"Fetched {len(steps)} sequence steps from HubSpot.") print(f"Fetched {len(dependencies)} sequence dependencies from HubSpot.") upsert_sequences(sequences, steps, dependencies) new_cursor_ms = max_updated_at_ms if max_updated_at_ms is not None else since_ms update_sync_metadata(supabase_client, "hubspot_sequences", now_iso) update_sync_metadata(supabase_client, "hubspot_sequence_steps", now_iso) update_sync_metadata(supabase_client, "hubspot_sequence_dependencies", now_iso) print(f"Sequences sync complete. Advanced cursor to {new_cursor_ms}.") logging.info("Sequences sync complete. Advanced cursor to %d.", new_cursor_ms) # ----------------------------------------------------------------------------- # CLI # ----------------------------------------------------------------------------- def _parse_cli_arg_to_ms(arg: str) -> int: """ Accept: - integer epoch ms - ISO-8601 (Z or offset) - YYYY-MM-DD (floors to 00:00Z) """ # epoch ms or seconds if re.fullmatch(r"\d{10,13}", arg): v = int(arg) if v < 10_000_000_000_000: # seconds -> ms v *= 1000 return v # YYYY-MM-DD if re.fullmatch(r"\d{4}-\d{2}-\d{2}", arg): d = datetime.datetime.strptime(arg, "%Y-%m-%d").replace(tzinfo=datetime.timezone.utc) return to_epoch_ms(floor_to_utc_midnight(d)) # ISO-8601 return to_epoch_ms(arg) if __name__ == "__main__": import sys if len(sys.argv) > 1: try: since = _parse_cli_arg_to_ms(sys.argv[1]) except Exception as e: print( f"Invalid timestamp. Provide epoch ms, ISO-8601, or YYYY-MM-DD. Error: {e}" ) sys.exit(1) main(since_ms=since) else: main()