Enhance Supabase utility functions to include affected row IDs. Edited texts in load data.

python/load_hubspot_data.py

@@ -145,16 +145,16 @@ def main():
         logging.exception("Error running emails pipeline: %s", e)
 
     try:
-        print("\n[6/7] Billing")
+        print("\n[6/7] Billing Services")
         hubspot_billing.main(since_ms=since_ms)
     except Exception as e:
-        logging.exception("Error running billing pipeline: %s", e)
+        logging.exception("Error running billing services pipeline: %s", e)
 
     try:
-        print("\n[7/7] Billing")
+        print("\n[7/7] Audit Logs")
         hubspot_audit.main(since_ms=since_ms)
     except Exception as e:
-        logging.exception("Error running billing pipeline: %s", e)
+        logging.exception("Error running audit log pipeline: %s", e)
 
     print("\n=== HubSpot sync complete ===")
 

python/supabase_utils.py

@@ -22,6 +22,41 @@ except Exception:
     _RETURN_MINIMAL = "minimal"
 
 
+def extract_row_ids(rows: List[Dict[str, Any]], on_conflict: Optional[List[str]] = None) -> List[Dict[str, Any]]:
+    """
+    Extract a list of dictionaries containing the on_conflict keys from each row.
+
+    If no on_conflict keys are provided, it will attempt to find a suitable key
+    by looking for the presence of "id", "uuid", "hubspot_id", or "object_id" in the first row.
+
+    If no conflict keys are found, it will return an empty list.
+
+    :param rows: List of dictionaries to extract IDs from
+    :param on_conflict: List of column names to use as conflict target
+    :return: List of dictionaries containing the on_conflict keys from each row
+    """
+    if not rows:
+        return []
+
+    keys: List[str] = list(on_conflict or [])
+
+    # Fallback if no explicit conflict keys were given
+    if not keys:
+        for candidate in ("id", "uuid", "hubspot_id", "object_id"):
+            if candidate in rows[0]:
+                keys = [candidate]
+                break
+
+    if not keys:
+        return []
+
+    id_list: List[Dict[str, Any]] = []
+    for r in rows:
+        id_list.append({k: r.get(k) for k in keys if k in r})
+
+    return id_list
+
+
 def insert_into_supabase_table(
     supabase_client: Any,
     table_name: str,
@@ -83,9 +118,13 @@ def insert_into_supabase_table(
             attempt += 1
 
             if attempt > max_retries:
+                affected_ids = extract_row_ids(rows, on_conflict)
                 logging.error(
-                    "Permanent failure upserting into %s after %d attempts: %s",
-                    table_name, attempt, e
+                    "Permanent failure upserting into %s after %d attempts: %s. Affected row ids: %s",
+                    table_name,
+                    attempt,
+                    e,
+                    affected_ids,
                 )
                 return False
 
@@ -105,9 +144,13 @@ def insert_into_supabase_table(
         except (httpx.RequestError, httpx.HTTPStatusError) as e:
             attempt += 1
             if attempt > max_retries:
+                affected_ids = extract_row_ids(rows, on_conflict)
                 logging.error(
-                    "Permanent HTTP failure upserting into %s after %d attempts: %s",
-                    table_name, attempt, e
+                    "Permanent HTTP failure upserting into %s after %d attempts: %s. Affected row ids: %s",
+                    table_name,
+                    attempt,
+                    e,
+                    affected_ids,
                 )
                 return False
 
@@ -121,9 +164,13 @@ def insert_into_supabase_table(
         except Exception as e:
             attempt += 1
             if attempt > max_retries:
+                affected_ids = extract_row_ids(rows, on_conflict)
                 logging.error(
-                    "Permanent unexpected failure upserting into %s after %d attempts: %s",
-                    table_name, attempt, e
+                    "Permanent unexpected failure upserting into %s after %d attempts: %s. Affected row ids: %s",
+                    table_name,
+                    attempt,
+                    e,
+                    affected_ids,
                 )
                 return False
 
@@ -214,9 +261,13 @@ def batched_insert(
                 # Other errors or batch already at min size -> retry with backoff
                 attempt += 1
                 if attempt > max_retries:
+                    affected_ids = extract_row_ids(rows, on_conflict)
                     logging.error(
-                        f"Failed inserting batch rows {i}–{i + current_batch_size} into {table_name} "
-                        f"after {max_retries} retries: {e}"
+                        "Permanent unexpected failure upserting into %s after %d attempts: %s. Affected row ids: %s",
+                        table_name,
+                        attempt,
+                        e,
+                        affected_ids,
                     )
                     raise