fix(warehouse-sources): skip HubSpot objects the grant cannot read

products/warehouse_sources/backend/temporal/data_imports/sources/hubspot/metadata.py

@@ -18,16 +18,17 @@
 from structlog.types import FilteringBoundLogger
 
 from products.warehouse_sources.backend.temporal.data_imports.sources.hubspot.helpers import fetch_data
+from products.warehouse_sources.backend.temporal.data_imports.sources.hubspot.scopes import HubspotForbiddenError
 from products.warehouse_sources.backend.temporal.data_imports.sources.hubspot.settings import (
     HUBSPOT_ENDPOINTS,
     HUBSPOT_METADATA_ENDPOINTS,
     PIPELINE_OBJECT_TYPES,
     apply_crm_api_version,
 )
 
-# A portal that hasn't enabled an object (or a grant that can't read it) answers with one of these
-# for that object type only. Skipping keeps one unavailable object type from failing the table.
-_SKIPPABLE_STATUSES = (403, 404)
+# A portal that hasn't enabled an object answers 404 for that object type only. An object the
+# grant cannot read answers 403, which `HubspotForbiddenError` carries instead.
+_SKIPPABLE_STATUSES = (404,)
 
 
 def _normalize_row(row: dict[str, Any], columns: list[str]) -> dict[str, Any]:
@@ -43,9 +44,18 @@ def _iter_pages(
     refresh_token: str,
     source_id: str | None,
     logger: FilteringBoundLogger,
+    *,
+    skip_forbidden: bool = False,
 ) -> Iterator[list[dict[str, Any]]]:
     try:
         yield from fetch_data(path, api_key, refresh_token, source_id=source_id)
+    except HubspotForbiddenError:
+        # A fan-out reads one path per object type, so an object the portal cannot read must not
+        # fail the whole table. A single-endpoint table keeps the 403: nothing is left to sync,
+        # and the customer must know to reconnect.
+        if not skip_forbidden:
+            raise
+        logger.warning(f"Hubspot: skipping {path} (status=403); the portal cannot read it")
     except requests.HTTPError as e:
         status = e.response.status_code if e.response is not None else None
         if status in _SKIPPABLE_STATUSES:
@@ -64,7 +74,7 @@ def get_pipelines_rows(
     columns = HUBSPOT_METADATA_ENDPOINTS["pipelines"].columns
     for object_type in PIPELINE_OBJECT_TYPES:
         path = apply_crm_api_version(f"/crm/v3/pipelines/{object_type}", api_version)
-        for page in _iter_pages(path, api_key, refresh_token, source_id, logger):
+        for page in _iter_pages(path, api_key, refresh_token, source_id, logger, skip_forbidden=True):
             # `stages` is dropped here — it is the pipeline_stages table.
             yield [
                 _normalize_row({**{k: v for k, v in p.items() if k != "stages"}, "object_type": object_type}, columns)
@@ -82,7 +92,7 @@ def get_pipeline_stages_rows(
     columns = HUBSPOT_METADATA_ENDPOINTS["pipeline_stages"].columns
     for object_type in PIPELINE_OBJECT_TYPES:
         path = apply_crm_api_version(f"/crm/v3/pipelines/{object_type}", api_version)
-        for page in _iter_pages(path, api_key, refresh_token, source_id, logger):
+        for page in _iter_pages(path, api_key, refresh_token, source_id, logger, skip_forbidden=True):
             rows = [
                 _normalize_row({**stage, "object_type": object_type, "pipeline_id": pipeline.get("id")}, columns)
                 for pipeline in page
@@ -102,7 +112,7 @@ def get_properties_rows(
     columns = HUBSPOT_METADATA_ENDPOINTS["properties"].columns
     for object_type in HUBSPOT_ENDPOINTS:
         path = apply_crm_api_version(f"/crm/v3/properties/{object_type}", api_version)
-        for page in _iter_pages(path, api_key, refresh_token, source_id, logger):
+        for page in _iter_pages(path, api_key, refresh_token, source_id, logger, skip_forbidden=True):
             yield [_normalize_row({**p, "object_type": object_type}, columns) for p in page]
 
 

products/warehouse_sources/backend/temporal/data_imports/sources/hubspot/test/test_hubspot_metadata.py

@@ -1,6 +1,7 @@
 import json
 from collections.abc import Iterator
 from typing import Any
+from urllib.parse import urlsplit
 
 import pytest
 from unittest.mock import MagicMock, patch
@@ -14,13 +15,18 @@
     get_pipelines_rows,
     get_properties_rows,
 )
+from products.warehouse_sources.backend.temporal.data_imports.sources.hubspot.scopes import HubspotForbiddenError
 from products.warehouse_sources.backend.temporal.data_imports.sources.hubspot.settings import (
     HUBSPOT_API_VERSION_2026_03,
     HUBSPOT_METADATA_ENDPOINTS,
     PIPELINE_OBJECT_TYPES,
 )
 
 _FETCH_DATA = "products.warehouse_sources.backend.temporal.data_imports.sources.hubspot.metadata.fetch_data"
+_SESSION = "products.warehouse_sources.backend.temporal.data_imports.sources.hubspot.helpers.make_tracked_session"
+
+PROPERTIES_PREFIX = "/crm/properties/2026-03"
+PIPELINES_PREFIX = "/crm/pipelines/2026-03"
 
 PIPELINE_PAYLOAD = [
     {
@@ -50,6 +56,20 @@ def _fake(path: str, *_args: Any, **_kwargs: Any) -> Iterator[list[dict[str, Any
     return _fake
 
 
+def _patch_session(forbidden: set[str], rows_by_path: dict[str, list[dict[str, Any]]]) -> Any:
+    # Patched at the HTTP boundary rather than at `fetch_data`, so the real status handling maps a
+    # 403 to the exception the fetchers have to cope with.
+    def _get(url: str, headers: Any = None, params: Any = None, timeout: Any = None) -> MagicMock:  # noqa: ARG001
+        path = urlsplit(url).path
+        response = MagicMock()
+        response.status_code = 403 if path in forbidden else 200
+        response.json.return_value = {"results": rows_by_path.get(path, [])}
+        return response
+
+    session = type("_S", (), {"get": staticmethod(_get)})()
+    return patch(_SESSION, new=lambda *_a, **_k: session)
+
+
 def _call(fetcher: Any) -> list[dict[str, Any]]:
     rows: list[dict[str, Any]] = []
     for page in fetcher(
@@ -138,22 +158,33 @@ def _fake(path: str, *_args: Any, **_kwargs: Any) -> Iterator[list[dict[str, Any
 
         assert captured == ["/crm/owners/2026-03"]
 
-    def test_properties_skips_an_object_type_the_portal_cannot_read(self) -> None:
-        # Property definitions are fanned out over every object endpoint, several of which need a
-        # scope the connection may not hold. One 403 must not take the whole table down.
-        response = MagicMock()
-        response.status_code = 403
-
-        def _fake(path: str, *_args: Any, **_kwargs: Any) -> Iterator[list[dict[str, Any]]]:
-            if path == "/crm/properties/2026-03/deals":
-                yield [{"name": "amount"}]
-                return
-            raise HTTPError("403 Client Error", response=response)
+    @pytest.mark.parametrize(
+        "fetcher,prefix,payload,forbidden_type,readable_type",
+        [
+            (get_properties_rows, PROPERTIES_PREFIX, [{"name": "amount"}], "feedback_submissions", "deals"),
+            (get_properties_rows, PROPERTIES_PREFIX, [{"name": "amount"}], "leads", "deals"),
+            (get_pipelines_rows, PIPELINES_PREFIX, PIPELINE_PAYLOAD, "tickets", "deals"),
+            (get_pipeline_stages_rows, PIPELINES_PREFIX, PIPELINE_PAYLOAD, "tickets", "deals"),
+        ],
+    )
+    def test_a_fan_out_skips_an_object_type_the_grant_cannot_read(
+        self,
+        fetcher: Any,
+        prefix: str,
+        payload: list[dict[str, Any]],
+        forbidden_type: str,
+        readable_type: str,
+    ) -> None:
+        # These tables are fanned out over every object endpoint, several of which need a scope the
+        # connection may not hold. "leads" is the scope-gated flavor, which raises a subclass.
+        with _patch_session({f"{prefix}/{forbidden_type}"}, {f"{prefix}/{readable_type}": payload}):
+            rows = _call(fetcher)
 
-        with patch(_FETCH_DATA, new=_fake):
-            rows = _call(get_properties_rows)
+        assert {r["object_type"] for r in rows} == {readable_type}
 
-        assert [r["name"] for r in rows] == ["amount"]
+    def test_owners_fails_the_table_when_the_grant_cannot_read_it(self) -> None:
+        with _patch_session({"/crm/owners/2026-03"}, {}), pytest.raises(HubspotForbiddenError):
+            _call(get_owners_rows)
 
     def test_server_error_still_fails_the_table(self) -> None:
         response = MagicMock()