✨Source HubSpot: New stream for contacts merged audit

airbyte-integrations/connectors/source-hubspot/README.md

@@ -36,6 +36,10 @@ The primary key for the following streams is `pipelineId`:
 
 - deal_pipelines
 
+The primary key for the following streams is `vid-to-merge`:
+
+- contacts_merged_audit
+
 The following streams do not have a primary key:
 
 - contact_lists (The primary key could potentially be a composite key (portalId, listId) - https://legacydocs.hubspot.com/docs/methods/lists/get_lists)

airbyte-integrations/connectors/source-hubspot/integration_tests/abnormal_state.json

@@ -43,6 +43,17 @@
       }
     }
   },
+  {
+    "type": "STREAM",
+    "stream": {
+      "stream_descriptor": {
+        "name": "contacts_merged_audit"
+      },
+      "stream_state": {
+        "updatedAt": "2221-10-12T13:37:56.412000+00:00"
+      }
+    }
+  },
   {
     "type": "STREAM",
     "stream": {

airbyte-integrations/connectors/source-hubspot/sample_files/basic_read_catalog.json

@@ -304,6 +304,15 @@
       },
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite"
+    },
+    {
+      "stream": {
+        "name": "contacts_merged_audit",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
     }
   ]
 }

airbyte-integrations/connectors/source-hubspot/sample_files/basic_read_oauth_catalog.json

@@ -208,6 +208,15 @@
       },
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite"
+    },
+    {
+      "stream": {
+        "name": "contacts_merged_audit",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
     }
   ]
 }

airbyte-integrations/connectors/source-hubspot/sample_files/full_refresh_catalog.json

@@ -45,6 +45,15 @@
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite"
     },
+    {
+      "stream": {
+        "name": "contacts_merged_audit",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    },
     {
       "stream": {
         "name": "deal_pipelines",

airbyte-integrations/connectors/source-hubspot/sample_files/full_refresh_oauth_catalog.json

@@ -197,6 +197,15 @@
       },
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite"
+    },
+    {
+      "stream": {
+        "name": "contacts_merged_audit",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
     }
   ]
 }

airbyte-integrations/connectors/source-hubspot/source_hubspot/schemas/contacts_merged_audit.json

@@ -0,0 +1,91 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": ["null", "object"],
+  "additionalProperties": true,
+  "properties": {
+    "canonical-vid": {
+      "type": ["null", "integer"]
+    },
+    "vid-to-merge": {
+      "type": ["null", "integer"]
+    },
+    "timestamp": {
+      "type": ["null", "integer"]
+    },
+    "entity-id": {
+      "type": ["null", "string"]
+    },
+    "user-id": {
+      "type": ["null", "integer"]
+    },
+    "num-properties-moved": {
+      "type": ["null", "integer"]
+    },
+    "merged_from_email": {
+      "type": ["null", "object"],
+      "additionalProperties": true,
+      "properties": {
+        "source-vids": {
+          "type": ["null", "array"],
+          "items": {
+            "type": ["null", "integer"]
+          }
+        },
+        "updated-by-user-id": {
+          "type": ["null", "integer"]
+        },
+        "source-label": {
+          "type": ["null", "string"]
+        },
+        "source-type": {
+          "type": ["null", "string"]
+        },
+        "value": {
+          "type": ["null", "string"]
+        },
+        "source-id": {
+          "type": ["null", "string"]
+        },
+        "selected": {
+          "type": ["null", "boolean"]
+        },
+        "timestamp": {
+          "type": ["null", "integer"]
+        }
+      }
+    },
+    "merged_to_email": {
+      "type": ["null", "object"],
+      "additionalProperties": true,
+      "properties": {
+        "updated-by-user-id": {
+          "type": ["null", "integer"]
+        },
+        "source-label": {
+          "type": ["null", "string"]
+        },
+        "source-type": {
+          "type": ["null", "string"]
+        },
+        "value": {
+          "type": ["null", "string"]
+        },
+        "source-id": {
+          "type": ["null", "string"]
+        },
+        "selected": {
+          "type": ["null", "boolean"]
+        },
+        "timestamp": {
+          "type": ["null", "integer"]
+        }
+      }
+    },
+    "first-name": {
+      "type": ["null", "string"]
+    },
+    "last-name": {
+      "type": ["null", "string"]
+    }
+  }
+}

airbyte-integrations/connectors/source-hubspot/source_hubspot/source.py

@@ -18,6 +18,7 @@
     ContactLists,
     Contacts,
     ContactsListMemberships,
+    ContactsMergedAudit,
     CustomObject,
     DealPipelines,
     Deals,
@@ -93,6 +94,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             ContactLists(**common_params),
             Contacts(**common_params),
             ContactsListMemberships(**common_params),
+            ContactsMergedAudit(**common_params),
             DealPipelines(**common_params),
             Deals(**common_params),
             DealsArchived(**common_params),

airbyte-integrations/connectors/source-hubspot/source_hubspot/streams.py

@@ -16,7 +16,7 @@
 import pendulum as pendulum
 import requests
 from airbyte_cdk.entrypoint import logger
-from airbyte_cdk.models import SyncMode
+from airbyte_cdk.models.airbyte_protocol import SyncMode
 from airbyte_cdk.sources import Source
 from airbyte_cdk.sources.streams import IncrementalMixin, Stream
 from airbyte_cdk.sources.streams.availability_strategy import AvailabilityStrategy
@@ -1775,6 +1775,74 @@ class Contacts(CRMSearchStream):
     scopes = {"crm.objects.contacts.read"}
 
 
+class ContactsMergedAudit(Stream):
+
+    url = "/contacts/v1/contact/vids/batch/"
+    updated_at_field = "timestamp"
+    scopes = {"crm.objects.contacts.read"}
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.config = kwargs
+
+    def get_json_schema(self) -> Mapping[str, Any]:
+        """Override get_json_schema defined in Stream class
+        Final object does not have properties field
+        We return JSON schema as defined in :
+        source_hubspot/schemas/contacts_merged_audit.json
+        """
+        return super(Stream, self).get_json_schema()
+
+    def stream_slices(
+        self, sync_mode: SyncMode, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None, **kwargs
+    ) -> Iterable[Mapping[str, Any]]:
+
+        slices = []
+
+        # we can query a max of 100 contacts at a time
+        max_contacts = 100
+        slices = []
+        contact_batch = []
+
+        contacts = Contacts(**self.config)
+        contacts._sync_mode = SyncMode.full_refresh
+        contacts.filter_old_records = False
+
+        for contact in contacts.read_records(sync_mode=SyncMode.full_refresh):
+            if contact["properties"].get("hs_merged_object_ids"):
+                contact_batch.append(contact["id"])
+
+                if len(contact_batch) == max_contacts:
+                    slices.append({"vid": contact_batch})
+                    contact_batch = []
+
+        if contact_batch:
+            slices.append({"vid": contact_batch})
+
+        return slices
+
+    def request_params(
+        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
+    ) -> MutableMapping[str, Any]:
+        return {"vid": stream_slice["vid"]}
+
+    def parse_response(
+        self,
+        response: requests.Response,
+        *,
+        stream_state: Mapping[str, Any],
+        stream_slice: Mapping[str, Any] = None,
+        next_page_token: Mapping[str, Any] = None,
+    ) -> Iterable[Mapping]:
+        response = self._parse_response(response)
+        if response.get("status", None) == "error":
+            self.logger.warning(f"Stream `{self.name}` cannot be procced. {response.get('message')}")
+            return
+
+        for contact_id in list(response.keys()):
+            yield from response[contact_id]["merge-audits"]
+
+
 class EngagementsCalls(CRMSearchStream):
     entity = "calls"
     last_modified_field = "hs_lastmodifieddate"

airbyte-integrations/connectors/source-hubspot/unit_tests/test_source.py

@@ -75,7 +75,7 @@ def test_streams(requests_mock, config):
 
     streams = SourceHubspot().streams(config)
 
-    assert len(streams) == 28
+    assert len(streams) == 29
 
 
 def test_check_credential_title_exception(config):