Clean Diff

tap_hubspot_beta/client_base.py

@@ -91,6 +91,24 @@ def request_records(self, context):
 
         while not finished:
             logging.getLogger("backoff").setLevel(logging.CRITICAL)
+
+            # only use companies stream for incremental syncs
+            if self.name == "companies":
+                fullsync_companies_state = self.tap_state.get("bookmarks", {}).get("fullsync_companies", {})
+                fullsync_on = False
+                try:
+                    # Check if the fullsync stream is selected or not
+                    fullsync_on = [s for s in self._tap.streams.items() if str(s[0]) == "fullsync_companies"][0][1].selected
+                except:
+                    pass
+                if fullsync_on and not fullsync_companies_state.get("replication_key") and self.is_first_sync():
+                    finished = True
+                    yield from []
+                    break
+                elif fullsync_companies_state.get("replication_key") and self.is_first_sync():
+                    self.stream_state.update(fullsync_companies_state)
+                    self.stream_state["starting_replication_value"] = self.stream_state["replication_key_value"]
+
             prepared_request = self.prepare_request(
                 context, next_page_token=next_page_token
             )
@@ -189,11 +207,23 @@ def schema(self):
         headers.update(self.authenticator.auth_headers or {})
         url = self.url_base + self.properties_url
         response = self.request_decorator(self.request_schema)(url, headers=headers)
-
         fields = response.json()
+
+        deduplicate_columns = self.config.get("deduplicate_columns", True)
+        base_properties = []
+        if isinstance(self.base_properties, list):
+            base_properties = [property.name.lower() for property in self.base_properties]
+
         for field in fields:
+            field_name = field.get("name")
+            # filter duplicated columns (case insensitive)
+            if deduplicate_columns:
+                if field_name.lower() in base_properties:
+                    self.logger.info(f"Not including field {field_name} in catalog as it's a duplicate(case insensitive) of a base property for stream {self.name}")
+                    continue
+
             if not field.get("deleted"):
-                property = th.Property(field.get("name"), self.extract_type(field))
+                property = th.Property(field_name, self.extract_type(field))
                 properties.append(property)
         return th.PropertiesList(*properties).to_dict()
 
@@ -308,6 +338,37 @@ def stream_maps(self) -> List[StreamMap]:
             ]
         return self._stream_maps
 
+    def process_row_types(self,row) -> Dict[str, Any]:
+        schema = self.schema['properties']
+        # If the row is null we ignore
+        if row is None:
+            return row
+
+        for field, value in row.items():
+            if field not in schema:
+                # Skip fields not found in the schema
+                continue
+
+            field_info = schema[field]
+            field_type = field_info.get("type", ["null"])[0]
+
+            if field_type == "boolean":
+                if value is None:
+                    row[field] = False
+                elif isinstance(value, str):
+                    # Attempt to cast to boolean
+                    if value.lower() == "true":
+                        row[field] = True
+                    elif value == "" or value.lower() == "false":
+                        row[field] = False
+
+        return row
+
+    def is_first_sync(self):
+        if self.stream_state.get("replication_key"):
+            return False
+        return True
+
 
 class hubspotStreamSchema(hubspotStream):
 

tap_hubspot_beta/client_v1.py

@@ -73,4 +73,5 @@ def post_process(self, row: dict, context: Optional[dict]) -> dict:
                         dt_field = datetime.fromtimestamp(int(row[field]) / 1000)
                         dt_field = dt_field.replace(tzinfo=None)
                         row[field] = dt_field.isoformat()
+        row = self.process_row_types(row)
         return row

tap_hubspot_beta/client_v2.py

@@ -0,0 +1,76 @@
+
+import logging
+from datetime import datetime
+from typing import Any, Dict, Optional, Iterable
+
+import requests
+from singer_sdk.helpers.jsonpath import extract_jsonpath
+
+from tap_hubspot_beta.client_base import hubspotStreamSchema
+import copy
+
+
+class hubspotV2Stream(hubspotStreamSchema):
+    """hubspot stream class."""
+
+    def get_url_params(
+        self, context: Optional[dict], next_page_token: Optional[Any]
+    ) -> Dict[str, Any]:
+        """Return a dictionary of values to be used in URL parameterization."""
+        params: dict = {}
+        params["limit"] = self.page_size
+        params.update(self.additional_prarams)
+        if self.properties_url:
+            params["properties"] = self.selected_properties
+        if next_page_token:
+            params["offset"] = next_page_token["offset"]
+        return params
+
+    def post_process(self, row: dict, context: Optional[dict]) -> dict:
+        """As needed, append or transform raw data to match expected structure."""
+        if self.properties_url:
+            for name, value in row.get("properties").items():
+                row[name] = value.get("value")
+            row["id"] = str(row["companyId"])
+            del row["properties"]
+        for field in self.datetime_fields:
+            if row.get(field) is not None:
+                if row.get(field) in [0, ""]:
+                    row[field] = None
+                else:
+                    # format datetime as hubspot standard ex. 2024-04-24T20:20:53.386Z
+                    dt_field = datetime.fromtimestamp(int(row[field]) / 1000)
+                    row[field] = dt_field.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+        row["updatedAt"] = row["hs_lastmodifieddate"]
+        row["createdAt"] = row["createdate"]
+        row["archived"] = False
+        row = self.process_row_types(row)
+        return row
+
+    def request_records(self, context: Optional[dict]) -> Iterable[dict]:
+        next_page_token: Any = None
+        finished = False
+        decorated_request = self.request_decorator(self._request)
+
+        while not finished:
+            prepared_request = self.prepare_request(
+                context, next_page_token=next_page_token
+            )
+            # only use fullsync_companies in the first sync
+            if self.name == "fullsync_companies" and not self.is_first_sync():
+                finished = True
+                yield from []
+                break
+            resp = decorated_request(prepared_request, context)
+            yield from self.parse_response(resp)
+            previous_token = copy.deepcopy(next_page_token)
+            next_page_token = self.get_next_page_token(
+                response=resp, previous_token=previous_token
+            )
+            if next_page_token and next_page_token == previous_token:
+                raise RuntimeError(
+                    f"Loop detected in pagination. "
+                    f"Pagination token {next_page_token} is identical to prior token."
+                )
+            # Cycle until get_next_page_token() no longer returns a value
+            finished = not next_page_token

tap_hubspot_beta/client_v3.py

@@ -8,6 +8,9 @@
 
 from tap_hubspot_beta.client_base import hubspotStream
 from pendulum import parse
+from singer_sdk import typing as th
+import singer
+
 
 from singer_sdk.exceptions import InvalidStreamSortException
 from singer_sdk.helpers._state import (
@@ -111,12 +114,16 @@ def post_process(self, row: dict, context: Optional[dict]) -> dict:
             for name, value in row["properties"].items():
                 row[name] = value
             del row["properties"]
+        # store archived value in _hg_archived
+        row["_hg_archived"] = False
+        row = self.process_row_types(row)
         return row
 
     def _sync_records(  # noqa C901  # too complex
         self, context: Optional[dict] = None
     ) -> None:
         """Sync records, emitting RECORD and STATE messages. """
+
         record_count = 0
         current_context: Optional[dict]
         context_list: Optional[List[dict]]
@@ -187,6 +194,16 @@ def _sync_records(  # noqa C901  # too complex
         # Reset interim bookmarks before emitting final STATE message:
         self._write_state_message()
 
+
+    def _sync_children(self, child_context: dict) -> None:
+        for child_stream in self.child_streams:
+            if child_stream.selected or child_stream.has_selected_descendents:
+                if not child_stream.bulk_child:
+                    ids = child_context.get("ids") or []
+                    for id in ids:
+                        child_stream.sync(context=id)
+                else:
+                    child_stream.sync(context=child_context)
 
 class hubspotV3Stream(hubspotStream):
     """hubspot stream class."""
@@ -209,7 +226,12 @@ def get_url_params(
         params["limit"] = self.page_size
         params.update(self.additional_prarams)
         if self.properties_url:
-            params["properties"] = ",".join(self.selected_properties)
+            # requesting either properties or properties with history
+            # if we send both it returns an error saying the url is too long
+            if params.get("propertiesWithHistory"):
+                params["propertiesWithHistory"] = ",".join(self.selected_properties)
+            else:
+                params["properties"] = ",".join(self.selected_properties)
         if next_page_token:
             params["after"] = next_page_token
         return params
@@ -220,6 +242,7 @@ def post_process(self, row: dict, context: Optional[dict]) -> dict:
             for name, value in row["properties"].items():
                 row[name] = value
             del row["properties"]
+        row = self.process_row_types(row)
         return row
 
 
@@ -275,4 +298,21 @@ def post_process(self, row: dict, context: Optional[dict]) -> dict:
             for name, value in row["properties"].items():
                 row[name] = value
             del row["properties"]
-        return row
+        row = self.process_row_types(row)
+        return row
+
+class hubspotHistoryV3Stream(hubspotV3Stream):
+
+    def post_process(self, row: dict, context) -> dict:
+        row = super().post_process(row, context)
+        props = row.get("propertiesWithHistory") or dict()
+        row["propertiesWithHistory"] = {k:v for (k,v) in props.items() if v}
+        row = {k:v for k,v in row.items() if k in ["id", "propertiesWithHistory", "createdAt", "updatedAt", "archived", "archivedAt"]}
+        return row
+
+    def _write_schema_message(self) -> None:
+        """Write out a SCHEMA message with the stream schema."""
+        for schema_message in self._generate_schema_messages():
+            schema_message.schema = th.PropertiesList(*self.base_properties).to_dict()
+            singer.write_message(schema_message)
+

tap_hubspot_beta/client_v4.py

@@ -19,6 +19,7 @@ class hubspotV4Stream(hubspotStream):
 
     rest_method = "POST"
     records_jsonpath = "$.results[*]"
+    bulk_child = True
 
     def get_url(self, context: Optional[dict]) -> str:
         """Get stream entity URL. """