Source Hubspot: fix "quotes" key error exception

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

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

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

@@ -3068,7 +3068,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-hubspot:0.1.38"
+- dockerImage: "airbyte/source-hubspot:0.1.39"
   spec:
     documentationUrl: "https://docs.airbyte.io/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.1.38
+LABEL io.airbyte.version=0.1.39
 LABEL io.airbyte.name=airbyte/source-hubspot

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

@@ -2,11 +2,25 @@
 # Copyright (c) 2021 Airbyte, Inc., all rights reserved.
 #
 
+import logging
+from typing import Any, MutableMapping
 
-from airbyte_cdk.sources.deprecated.base_source import BaseSource
+from airbyte_cdk.sources.deprecated.base_source import BaseClient, BaseSource, ConfiguredAirbyteStream
 
 from .client import Client
 
 
 class SourceHubspot(BaseSource):
     client_class = Client
+
+    def _read_stream(
+        self, logger: logging.Logger, client: BaseClient, configured_stream: ConfiguredAirbyteStream, state: MutableMapping[str, Any]
+    ):
+        """
+        This method is overridden to check if the stream exists in the client.
+        """
+        stream_name = configured_stream.stream.name
+        if not client._apis.get(stream_name):
+            logger.warning(f"Stream {stream_name} does not exist in the client.")
+            return
+        yield from super()._read_stream(logger=logger, client=client, configured_stream=configured_stream, state=state)

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

@@ -3,14 +3,19 @@
 #
 
 
+import logging
 from functools import partial
 
 import pytest
+from airbyte_cdk.sources.deprecated.base_source import ConfiguredAirbyteCatalog, Type
 from source_hubspot.api import API, PROPERTIES_PARAM_MAX_LENGTH, split_properties
 from source_hubspot.client import Client
+from source_hubspot.source import SourceHubspot
 
 NUMBER_OF_PROPERTIES = 2000
 
+logger = logging.getLogger("test_client")
+
 
 @pytest.fixture(name="some_credentials")
 def some_credentials_fixture():
@@ -260,3 +265,51 @@ def test_stream_with_splitting_properties_with_new_record(self, requests_mock, c
         stream_records = list(test_stream.read(getter=partial(self.get, test_stream.url, api=api)))
 
         assert len(stream_records) == 6
+
+
+@pytest.fixture(name="oauth_config")
+def oauth_config_fixture():
+    return {
+        "start_date": "2021-10-10T00:00:00Z",
+        "credentials": {
+            "credentials_title": "OAuth Credentials",
+            "redirect_uri": "https://airbyte.io",
+            "client_id": "test_client_id",
+            "client_secret": "test_client_secret",
+            "refresh_token": "test_refresh_token",
+            "access_token": "test_access_token",
+            "token_expires": "2021-05-30T06:00:00Z",
+        },
+    }
+
+
+@pytest.fixture(name="configured_catalog")
+def configured_catalog_fixture():
+    configured_catalog = {
+        "streams": [
+            {
+                "stream": {
+                    "name": "quotes",
+                    "json_schema": {},
+                    "supported_sync_modes": ["full_refresh", "incremental"],
+                    "source_defined_cursor": True,
+                    "default_cursor_field": ["updatedAt"],
+                },
+                "sync_mode": "incremental",
+                "cursor_field": ["updatedAt"],
+                "destination_sync_mode": "append",
+            }
+        ]
+    }
+    return ConfiguredAirbyteCatalog.parse_obj(configured_catalog)
+
+
+def test_it_should_not_read_quotes_stream_if_it_does_not_exist_in_client(oauth_config, configured_catalog):
+    """
+    If 'quotes' stream is not in the client, it should skip it.
+    """
+    source = SourceHubspot()
+
+    all_records = list(source.read(logger, config=oauth_config, catalog=configured_catalog, state=None))
+    records = [record for record in all_records if record.type == Type.RECORD]
+    assert not records

docs/integrations/sources/hubspot.md

@@ -112,6 +112,7 @@ If you are using Oauth, most of the streams require the appropriate [scopes](htt
 
 | Version | Date       | Pull Request | Subject                                                                                                                                        |
 |:--------|:-----------| :--- |:-----------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.1.39  | 2022-02-10 | [10055](https://github.com/airbytehq/airbyte/pull/10055) | Bug fix: reading not initialized stream |
 | 0.1.38  | 2022-02-03 | [9786](https://github.com/airbytehq/airbyte/pull/9786) | Add new streams for engagements(calls, emails, meetings, notes and tasks) |
 | 0.1.37  | 2022-01-27 | [9555](https://github.com/airbytehq/airbyte/pull/9555) | Getting form_submission for all forms |
 | 0.1.36  | 2022-01-22 | [7784](https://github.com/airbytehq/airbyte/pull/7784) | Add Property History Stream |