Source Hubspot: fix "quotes" key error exception

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