diff --git a/products/warehouse_sources/backend/temporal/data_imports/sources/hubspot/metadata.py b/products/warehouse_sources/backend/temporal/data_imports/sources/hubspot/metadata.py index 418a08a34da4..97ec33832254 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/sources/hubspot/metadata.py +++ b/products/warehouse_sources/backend/temporal/data_imports/sources/hubspot/metadata.py @@ -18,6 +18,7 @@ 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, @@ -25,9 +26,9 @@ 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] diff --git a/products/warehouse_sources/backend/temporal/data_imports/sources/hubspot/test/test_hubspot_metadata.py b/products/warehouse_sources/backend/temporal/data_imports/sources/hubspot/test/test_hubspot_metadata.py index e7599239710a..96c679d7c115 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/sources/hubspot/test/test_hubspot_metadata.py +++ b/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,6 +15,7 @@ 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, @@ -21,6 +23,10 @@ ) _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()