🎉Source Hubspot: Add contacts associations to Deals stream.

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/36c891d9-4bd9-43ac-bad2-10e12756272c.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "36c891d9-4bd9-43ac-bad2-10e12756272c",
   "name": "Hubspot",
   "dockerRepository": "airbyte/source-hubspot",
-  "dockerImageTag": "0.1.10",
+  "dockerImageTag": "0.1.11",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/hubspot",
   "icon": "hubspot.svg"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -137,7 +137,7 @@
 - sourceDefinitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c
   name: Hubspot
   dockerRepository: airbyte/source-hubspot
-  dockerImageTag: 0.1.10
+  dockerImageTag: 0.1.11
   documentationUrl: https://docs.airbyte.io/integrations/sources/hubspot
   icon: hubspot.svg
 - sourceDefinitionId: 95e8cffd-b8c4-4039-968e-d32fb4a69bde

airbyte-integrations/connectors/source-hubspot/Dockerfile

@@ -14,5 +14,5 @@ RUN pip install .
 
 ENV AIRBYTE_ENTRYPOINT "/airbyte/base.sh"
 
-LABEL io.airbyte.version=0.1.10
+LABEL io.airbyte.version=0.1.11
 LABEL io.airbyte.name=airbyte/source-hubspot

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

@@ -620,6 +620,63 @@ def list(self, fields) -> Iterable:
             yield record
 
 
+class CRMAssociationStream(Stream):
+    """  see https://developers.hubspot.com/docs/api/crm/associations for more details
+    """
+
+    entity: Optional[str] = None
+    associations: List[str] = []
+    updated_at_field = None
+    created_at_field = None
+
+    @property
+    def url(self):
+        """Entity URL"""
+        return f"/crm/v3/associations/{self._relationship_from}/{self._relationship_to}/{self._endpoint}"
+
+    def __init__(self, relationship_from: str = None, relationship_to: str = None, endpoint: str = None, **kwargs):
+        super().__init__(**kwargs)
+        self._relationship_from = relationship_from
+        self._relationship_to = relationship_to
+        self._endpoint = endpoint
+
+    def _transform(self, records: Iterable) -> Iterable:
+        """Preprocess record """
+        for record in records:
+            yield record
+
+    def _filter_old_records(self, records: Iterable) -> Iterable:
+        """Skip """
+        for record in records:
+            yield record
+
+
+class DealToContactAssociationsStream(CRMAssociationStream):
+    def __init__(self, deal_stream, **kwargs):
+        super().__init__(relationship_from="Deals", relationship_to="Contacts", endpoint="batch/read", **kwargs)
+        self._deal_stream = deal_stream
+
+    def prepare_payload(self, ids: Iterable) -> Mapping[str, Any]:
+        payload = {"inputs": []}
+
+        for deal_id in ids:
+            payload['inputs'].append({"id": deal_id})
+
+        return payload
+
+    def list(self, fields) -> Iterable:
+
+        deals = self._deal_stream.list(fields)
+        ids = []
+
+        for deal in deals:
+            ids.append(deal.get('id'))
+            if len(ids) == self.limit:
+                payload = self.prepare_payload(ids=ids)
+                yield from self.read(partial(self._api.post, url=self.url, data=payload))
+                ids = []
+
+
 class DealPipelineStream(Stream):
     """Deal pipelines, API v1,
     This endpoint requires the contacts scope the tickets scope.

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

@@ -41,6 +41,7 @@
     OwnerStream,
     SubscriptionChangeStream,
     WorkflowStream,
+    DealToContactAssociationsStream,
 )
 
 
@@ -69,6 +70,7 @@ def __init__(self, start_date, credentials, **kwargs):
             "subscription_changes": SubscriptionChangeStream(**common_params),
             "tickets": CRMObjectStream(entity="ticket", **common_params),
             "workflows": WorkflowStream(**common_params),
+            "deal_to_contact_associations": DealToContactAssociationsStream(deal_stream=DealStream(**common_params), **common_params),
         }
 
         super().__init__(**kwargs)

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

@@ -0,0 +1,24 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": ["null", "object"],
+  "properties": {
+    "from": {
+      "type": ["null", "Object"],
+      "properties": {
+        "id": {
+          "type": ["null", "string"]    
+        }
+      }
+    },
+    "to": {
+      "type": ["null", "array"],
+      "items": {
+        "type": ["null", "Object"],
+        "properties":{
+          "id": ["null", "string"],
+          "type": ["null", "string"]
+        }
+      }
+    }
+  }
+}