🐛 Source HubSpot: fix property scopes (#18624)

* 🐛 Source HubSpot: fix property scopes

* 🐛 Source HubSpot: bump version

* 🐛 Source Hubspot: fix properties for auth

* 🐛 Source Hubspot: fix log messages

* auto-bump connector version

Co-authored-by: Octavia Squidington III <[email protected]>

Author: artem1205

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

@@ -572,7 +572,7 @@
 - name: HubSpot
   sourceDefinitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c
   dockerRepository: airbyte/source-hubspot
-  dockerImageTag: 0.2.2
+  dockerImageTag: 0.2.3
   documentationUrl: https://docs.airbyte.com/integrations/sources/hubspot
   icon: hubspot.svg
   sourceType: api

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

@@ -5324,7 +5324,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-hubspot:0.2.2"
+- dockerImage: "airbyte/source-hubspot:0.2.3"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/hubspot"
     connectionSpecification:

airbyte-integrations/connectors/source-hubspot/Dockerfile

@@ -34,5 +34,5 @@ COPY source_hubspot ./source_hubspot
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.2.2
+LABEL io.airbyte.version=0.2.3
 LABEL io.airbyte.name=airbyte/source-hubspot

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

@@ -3,6 +3,7 @@
 #
 
 import logging
+from itertools import chain
 from typing import Any, Iterator, List, Mapping, MutableMapping, Optional, Tuple, Union
 
 import requests
@@ -130,6 +131,12 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             available_streams = [stream for stream in streams if stream.scope_is_granted(granted_scopes)]
             unavailable_streams = [stream for stream in streams if not stream.scope_is_granted(granted_scopes)]
             self.logger.info(f"The following streams are unavailable: {[s.name for s in unavailable_streams]}")
+            partially_available_streams = [stream for stream in streams if not stream.properties_scope_is_granted()]
+            required_scoped = set(chain(*[x.properties_scopes for x in partially_available_streams]))
+            self.logger.info(
+                f"The following streams are partially available: {[s.name for s in partially_available_streams]}, "
+                f"add the following scopes to download all available data: {required_scoped}"
+            )
         else:
             self.logger.info("No scopes to grant when authenticating with API key.")
             available_streams = streams

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

@@ -209,13 +209,16 @@ class Stream(HttpStream, ABC):
     filter_old_records: bool = True
     denormalize_records: bool = False  # one record from API response can result in multiple records emitted
     raise_on_http_errors: bool = True
+    granted_scopes: Set = None
+    properties_scopes: Set = None
 
     @property
     @abstractmethod
     def scopes(self) -> Set[str]:
         """Set of required scopes. Users need to grant at least one of the scopes for the stream to be avaialble to them"""
 
     def scope_is_granted(self, granted_scopes: Set[str]) -> bool:
+        self.granted_scopes = set(granted_scopes)
         if not self.scopes:
             return True
         else:
@@ -631,16 +634,24 @@ def _get_field_props(field_type: str) -> Mapping[str, List[str]]:
     @lru_cache()
     def properties(self) -> Mapping[str, Any]:
         """Some entities has dynamic set of properties, so we trying to resolve those at runtime"""
-        if not self.entity:
-            return {}
-
         props = {}
+        if not self.entity:
+            return props
+        if not self.properties_scope_is_granted():
+            logger.warning(
+                f"Check your API key has the following permissions granted: {self.properties_scopes}, "
+                f"to be able to fetch all properties available."
+            )
+            return props
         data, response = self._api.get(f"/properties/v2/{self.entity}/properties")
         for row in data:
             props[row["name"]] = self._get_field_props(row["type"])
 
         return props
 
+    def properties_scope_is_granted(self):
+        return not self.properties_scopes - self.granted_scopes if self.properties_scopes and self.granted_scopes else True
+
     def _flat_associations(self, records: Iterable[MutableMapping]) -> Iterable[MutableMapping]:
         """When result has associations we prefer to have it flat, so we transform this:
 
@@ -1127,6 +1138,7 @@ class ContactsListMemberships(Stream):
     page_field = "vid-offset"
     primary_key = "canonical-vid"
     scopes = {"crm.objects.contacts.read"}
+    properties_scopes = {"crm.schemas.contacts.read"}
 
     def _transform(self, records: Iterable) -> Iterable:
         """Extracting list membership records from contacts
@@ -1421,6 +1433,7 @@ class PropertyHistory(Stream):
     limit_field = "count"
     limit = 100
     scopes = {"crm.objects.contacts.read"}
+    properties_scopes = {"crm.schemas.contacts.read"}
 
     def request_params(
         self,