Source Hubspot: handle 400/403 errors with logger warnings (#29211)

* Source Hubspot: handle 400/403 errors with logger warnings

* Update changelog

* Update expected records

* Make changes accordingly to PR review

* Fix format errors

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=1.3.0
+LABEL io.airbyte.version=1.3.1
 LABEL io.airbyte.name=airbyte/source-hubspot

airbyte-integrations/connectors/source-hubspot/metadata.yaml

@@ -5,7 +5,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c
-  dockerImageTag: 1.2.0
+  dockerImageTag: 1.3.1
   dockerRepository: airbyte/source-hubspot
   githubIssueLabel: source-hubspot
   icon: hubspot.svg
@@ -14,7 +14,7 @@ data:
   registries:
     cloud:
       enabled: true
-      dockerImageTag: 1.2.0
+      dockerImageTag: 1.3.1
     oss:
       enabled: true
   releaseStage: generally_available

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

@@ -45,7 +45,7 @@ class HubspotAccessDenied(HubspotError):
     """403 Forbidden"""
 
 
-class HubspotRateLimited(HubspotError):
+class HubspotRateLimited(HTTPError):
     """429 Rate Limit Reached"""
 
 

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

@@ -29,14 +29,7 @@ from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer
 from airbyte_cdk.utils import AirbyteTracedException
 from requests import HTTPError, codes
 from source_hubspot.constants import OAUTH_CREDENTIALS, PRIVATE_APP_CREDENTIALS
-from source_hubspot.errors import (
-    HubspotAccessDenied,
-    HubspotBadRequest,
-    HubspotInvalidAuth,
-    HubspotRateLimited,
-    HubspotTimeout,
-    InvalidStartDateConfigError,
-)
+from source_hubspot.errors import HubspotAccessDenied, HubspotInvalidAuth, HubspotRateLimited, HubspotTimeout, InvalidStartDateConfigError
 from source_hubspot.helpers import APIv1Property, APIv3Property, GroupByKey, IRecordPostProcessor, IURLPropertyRepresentation, StoreAsIs
 
 # we got this when provided API Token has incorrect format
@@ -194,10 +187,10 @@ class API:
 
         if response.status_code == HTTPStatus.BAD_REQUEST:
             message = f"Request to {response.url} didn't succeed. Please verify your credentials and try again.\nError message from Hubspot API: {message}"
-            raise HubspotBadRequest(internal_message=message, failure_type=FailureType.config_error, response=response)
+            logger.warning(message)
         elif response.status_code == HTTPStatus.FORBIDDEN:
             message = f"The authenticated user does not have permissions to access the URL {response.url}. Verify your permissions to access this endpoint."
-            raise HubspotAccessDenied(internal_message=message, failure_type=FailureType.config_error, response=response)
+            logger.warning(message)
         elif response.status_code in (HTTPStatus.UNAUTHORIZED, CLOUDFLARE_ORIGIN_DNS_ERROR):
             message = (
                 "The user cannot be authorized with provided credentials. Please verify that your credentails are valid and try again."
@@ -205,10 +198,9 @@ class API:
             raise HubspotInvalidAuth(internal_message=message, failure_type=FailureType.config_error, response=response)
         elif response.status_code == HTTPStatus.TOO_MANY_REQUESTS:
             retry_after = response.headers.get("Retry-After")
+            message = f"You have reached your Hubspot API limit. We will resume replication once after {retry_after} seconds.\nSee https://developers.hubspot.com/docs/api/usage-details"
             raise HubspotRateLimited(
-                internal_message=f"You have reached your Hubspot API limit. We will resume replication once after {retry_after} seconds."
-                " See https://developers.hubspot.com/docs/api/usage-details",
-                failure_type=FailureType.config_error,
+                message,
                 response=response,
             )
         elif response.status_code in (HTTPStatus.BAD_GATEWAY, HTTPStatus.SERVICE_UNAVAILABLE):

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

@@ -12,7 +12,7 @@ import mock
 import pendulum
 import pytest
 from airbyte_cdk.models import ConfiguredAirbyteCatalog, SyncMode, Type
-from source_hubspot.errors import HubspotAccessDenied, HubspotRateLimited, InvalidStartDateConfigError
+from source_hubspot.errors import HubspotRateLimited, InvalidStartDateConfigError
 from source_hubspot.helpers import APIv3Property
 from source_hubspot.source import SourceHubspot
 from source_hubspot.streams import API, Companies, Deals, Engagements, MarketingEmails, Products, Stream
@@ -179,11 +179,11 @@ def test_stream_forbidden(requests_mock, config, caplog):
         }
     )
 
-    with pytest.raises(HubspotAccessDenied):
-        records = list(SourceHubspot().read(logger, config, catalog, {}))
-        assert json["message"] in caplog.text
-        records = [r for r in records if r.type == Type.RECORD]
-        assert not records
+    records = list(SourceHubspot().read(logger, config, catalog, {}))
+    assert json["message"] in caplog.text
+    assert "The authenticated user does not have permissions to access the URL" in caplog.text
+    records = [r for r in records if r.type == Type.RECORD]
+    assert not records
 
 
 def test_parent_stream_forbidden(requests_mock, config, caplog, fake_properties_list):
@@ -220,11 +220,11 @@ def test_parent_stream_forbidden(requests_mock, config, caplog, fake_properties_
         }
     )
 
-    with pytest.raises(HubspotAccessDenied):
-        records = list(SourceHubspot().read(logger, config, catalog, {}))
-        assert json["message"] in caplog.text
-        records = [r for r in records if r.type == Type.RECORD]
-        assert not records
+    records = list(SourceHubspot().read(logger, config, catalog, {}))
+    assert json["message"] in caplog.text
+    assert "The authenticated user does not have permissions to access the URL" in caplog.text
+    records = [r for r in records if r.type == Type.RECORD]
+    assert not records
 
 
 class TestSplittingPropertiesFunctionality: