From f6a566458ff83cea9d84d6eae3260b030d346f57 Mon Sep 17 00:00:00 2001 From: Vladimir Klimontovich Date: Wed, 1 Jul 2026 14:47:24 +0300 Subject: [PATCH] feat(source-hubspot): capture CRM deletions as in-stream tombstones The POST /crm/v3/objects//search endpoint never returns archived records, so deletions are invisible on incremental syncs. Add a HubspotDeletionRetriever that runs the normal live search pass plus a second GET /crm/v3/objects/?archived=true pass, emitting archived objects into the same stream as tombstones carrying the same PK id, full properties_* columns, _ab_cdc_deleted_at = archivedAt (fallback createdAt), and updatedAt = archivedAt. Wired for companies, contacts, deals, leads, engagements_calls, engagements_emails, engagements_notes, engagements_tasks. Refs #47198, #40595 --- .../connectors/source-hubspot/components.py | 112 ++++++++++ .../connectors/source-hubspot/manifest.yaml | 204 +++++++++++++++++- .../connectors/source-hubspot/metadata.yaml | 2 +- .../test_hubspot_deletion_retriever.py | 64 ++++++ docs/integrations/sources/hubspot.md | 1 + 5 files changed, 374 insertions(+), 9 deletions(-) create mode 100644 airbyte-integrations/connectors/source-hubspot/unit_tests/test_hubspot_deletion_retriever.py diff --git a/airbyte-integrations/connectors/source-hubspot/components.py b/airbyte-integrations/connectors/source-hubspot/components.py index 4455edf47373..b9a9ffc7b5ef 100644 --- a/airbyte-integrations/connectors/source-hubspot/components.py +++ b/airbyte-integrations/connectors/source-hubspot/components.py @@ -3,6 +3,7 @@ # import logging +import threading from dataclasses import InitVar, dataclass, field from datetime import timedelta from typing import Any, Dict, Iterable, List, Mapping, MutableMapping, Optional, Union @@ -37,6 +38,7 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import PaginationStrategy from airbyte_cdk.sources.declarative.requesters.request_options import InterpolatedRequestOptionsProvider from airbyte_cdk.sources.declarative.requesters.requester import Requester +from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever from airbyte_cdk.sources.declarative.schema.schema_loader import SchemaLoader from airbyte_cdk.sources.declarative.transformations import RecordTransformation from airbyte_cdk.sources.streams.http.error_handlers.response_models import ErrorResolution, ResponseAction @@ -782,6 +784,116 @@ def build_associations_retriever( ) +@dataclass +class HubspotDeletionRetriever(Retriever): + """ + Retriever for CRM Search streams (e.g. `companies`, `deals`, `contacts`) that, in addition + to the live records returned by the `POST /crm/v3/objects//search` endpoint, emits + tombstone records for ARCHIVED (deleted) objects fetched from the CRM list endpoint + (`GET /crm/v3/objects/?archived=true`). + + The `/search` endpoint never returns archived records, so deleted objects are otherwise + invisible on incremental syncs. Each archived object is emitted once per sync as a tombstone + that carries the same primary key `id` and the SAME full set of flattened `properties_*` + columns as a live record (so the destination's merge does not NULL-overwrite those columns), + plus: + - `_ab_cdc_deleted_at` = archivedAt (fallback createdAt) + - `updatedAt` = archivedAt (so the tombstone sorts after the live row for + downstream primary-key dedup) + + Design + ------ + This is a thin wrapper that delegates the live read to a fully-built nested `live_retriever` + (a `SimpleRetriever` created through the normal factory so it keeps all of its wiring: + query-property fetching/chunking, decoder, pagination, cursor, etc.) and performs an extra + archived pass through a nested `archived_retriever`. Subclassing `SimpleRetriever` directly + and swapping the manifest `type` was rejected because the generic custom-component builder + bypasses `create_simple_retriever` and drops that wiring. + + Single emission + --------------- + This source runs on the concurrent path (ConcurrentDeclarativeSource). There, one shared + retriever instance serves every partition (time slice) and `read_records` is called once per + slice. A lock-guarded flag guarantees the archived pass runs exactly once per sync, regardless + of how many live slices/threads are in play. + """ + + live_retriever: Retriever + config: Config + parameters: InitVar[Mapping[str, Any]] + archived_retriever: Optional[Retriever] = None + deletion_marker_field: str = "_ab_cdc_deleted_at" + deletion_cursor_field: str = "archivedAt" + + def __post_init__(self, parameters: Mapping[str, Any]) -> None: + self._parameters = parameters + self._archived_lock = threading.Lock() + self._archived_emitted = False + # Reused to normalize archived records against the (dynamic) stream schema, exactly + # like the live record selector does for live records. + self._entity_normalizer = EntitySchemaNormalization() + + def read_records( + self, + records_schema: Mapping[str, Any], + stream_slice: Optional[StreamSlice] = None, + ) -> Iterable[Any]: + # 1) Live records for this slice (unchanged behavior). + yield from self.live_retriever.read_records(records_schema, stream_slice=stream_slice) + + # 2) Archived (tombstone) records: emit exactly once per sync, on whichever slice + # wins the flag first. + if self.archived_retriever is None: + return + with self._archived_lock: + if self._archived_emitted: + return + self._archived_emitted = True + # The archived list endpoint is not available for every object type / scope (e.g. + # HubSpot returns 403 or "Paging through deleted objects is not yet supported" for + # some objects). Never let that fail the live sync: log and continue with no tombstones. + try: + yield from self._read_archived_records(records_schema) + except Exception as exc: # noqa: BLE001 - deletion capture is best-effort + logger.warning( + "Archived (deletion) pass failed; continuing without tombstones. Error: %s", + exc, + ) + + def _read_archived_records(self, records_schema: Mapping[str, Any]) -> Iterable[Any]: + # A single empty slice: the archived retriever paginates the whole archived set. + archived_slice = StreamSlice(partition={}, cursor_slice={}) + for item in self.archived_retriever.read_records(records_schema, stream_slice=archived_slice): + data = item.data if isinstance(item, Record) else item + if not isinstance(data, Mapping): + # Non-record message (log/state/etc.) - pass through untouched. + yield item + continue + yield self._to_tombstone(dict(data), records_schema) + + def _to_tombstone(self, data: Dict[str, Any], records_schema: Mapping[str, Any]) -> Dict[str, Any]: + # Flatten `properties` -> `properties_*` (idempotent; mirrors the live DpathFlattenFields + # with delete_origin_value=false so the raw `properties` object is preserved too). + self._flatten_properties(data) + + # Backfill the deletion cursor from archivedAt (fallback createdAt) and tag the record. + deleted_at = data.get(self.deletion_cursor_field) or data.get("createdAt") + data[self.deletion_cursor_field] = deleted_at + data[self.deletion_marker_field] = deleted_at + data["updatedAt"] = deleted_at + + # Normalize types/datetimes against the stream schema, same as live records. + self._entity_normalizer.transform(data, records_schema) + return data + + @staticmethod + def _flatten_properties(data: Dict[str, Any]) -> None: + properties = data.get("properties") + if isinstance(properties, Mapping): + for key, value in properties.items(): + data[f"properties_{key}"] = value + + @dataclass class HubspotCRMSearchPaginationStrategy(PaginationStrategy): """ diff --git a/airbyte-integrations/connectors/source-hubspot/manifest.yaml b/airbyte-integrations/connectors/source-hubspot/manifest.yaml index d9159236992a..278179830765 100644 --- a/airbyte-integrations/connectors/source-hubspot/manifest.yaml +++ b/airbyte-integrations/connectors/source-hubspot/manifest.yaml @@ -750,7 +750,7 @@ definitions: $ref: "#/schemas/email_subscriptions" deals_stream: - $ref: "#/definitions/base_crm_search_incremental_stream" + $ref: "#/definitions/base_crm_search_incremental_deletion_stream" transformations: - type: CustomTransformation class_name: source_declarative_manifest.components.NewtoLegacyFieldTransformation @@ -869,6 +869,7 @@ definitions: schema: $ref: "#/schemas/deals_archived" + forms_stream: primary_key: - id @@ -1662,6 +1663,65 @@ definitions: cursor_granularity: PT0.001S lookback_window: PT{{ config.get('lookback_window', 0) }}M + # Same as base_crm_search_incremental_stream, but the retriever also emits ARCHIVED + # (deleted) objects as tombstone records in the same stream (see HubspotDeletionRetriever). + # A stream opts in by `$ref`-ing this definition instead of base_crm_search_incremental_stream. + # + # The archived list endpoint is /crm/v3/objects/?archived=true and the property + # list is /properties/v2//properties. For all currently-wired objects the search + # `entity` param doubles as the object type, so it is the default; a stream can override with + # `archived_object_path` / `archived_properties_object` in its $parameters if they differ. + base_crm_search_incremental_deletion_stream: + $ref: "#/definitions/base_crm_search_incremental_stream" + retriever: + type: CustomRetriever + class_name: source_declarative_manifest.components.HubspotDeletionRetriever + # Live behavior is exactly base_crm_search_incremental_stream's retriever. + live_retriever: + $ref: "#/definitions/base_crm_search_incremental_stream/retriever" + # Archived companies/contacts/deals/... : LIST endpoint with archived=true, fetching the + # full property set so tombstones carry the same properties_* columns as live records. + archived_retriever: + type: SimpleRetriever + requester: + $ref: "#/definitions/base_requester" + path: "/crm/v3/objects/{{ parameters.get('archived_object_path') or parameters['entity'] }}" + request_parameters: + archived: "true" + properties: + type: QueryProperties + property_list: + type: PropertiesFromEndpoint + property_field_path: + - name + retriever: + type: SimpleRetriever + requester: + $ref: "#/definitions/base_requester" + path: "/properties/v2/{{ parameters.get('archived_properties_object') or parameters['entity'] }}/properties" + decoder: + type: JsonDecoder + record_selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: [] + property_chunking: + type: PropertyChunking + property_limit_type: characters + property_limit: 15000 + record_merge_strategy: + type: GroupByKeyMergeStrategy + key: id + paginator: + $ref: "#/definitions/cursor_paginator" + record_selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: + - results + base_crm_search_incremental_parent_stream: primary_key: - id @@ -1717,7 +1777,7 @@ definitions: lookback_window: PT{{ config.get('lookback_window', 0) }}M companies_stream: - $ref: "#/definitions/base_crm_search_incremental_stream" + $ref: "#/definitions/base_crm_search_incremental_deletion_stream" $parameters: name: companies entity: company @@ -1732,7 +1792,7 @@ definitions: - $ref: "#/definitions/base_dynamic_schema_loader" engagements_calls_stream: - $ref: "#/definitions/base_crm_search_incremental_stream" + $ref: "#/definitions/base_crm_search_incremental_deletion_stream" $parameters: name: engagements_calls entity: calls @@ -1750,7 +1810,7 @@ definitions: - $ref: "#/definitions/base_dynamic_schema_loader" engagements_emails_stream: - $ref: "#/definitions/base_crm_search_incremental_stream" + $ref: "#/definitions/base_crm_search_incremental_deletion_stream" $parameters: name: engagements_emails entity: emails @@ -1786,7 +1846,7 @@ definitions: - $ref: "#/definitions/base_dynamic_schema_loader" engagements_notes_stream: - $ref: "#/definitions/base_crm_search_incremental_stream" + $ref: "#/definitions/base_crm_search_incremental_deletion_stream" $parameters: name: engagements_notes entity: notes @@ -1804,7 +1864,7 @@ definitions: - $ref: "#/definitions/base_dynamic_schema_loader" engagements_tasks_stream: - $ref: "#/definitions/base_crm_search_incremental_stream" + $ref: "#/definitions/base_crm_search_incremental_deletion_stream" $parameters: name: engagements_tasks entity: tasks @@ -2044,7 +2104,7 @@ definitions: $ref: "#/schemas/line_items" contacts_stream: - $ref: "#/definitions/base_crm_search_incremental_stream" + $ref: "#/definitions/base_crm_search_incremental_deletion_stream" transformations: - type: CustomTransformation class_name: source_declarative_manifest.components.NewtoLegacyFieldTransformation @@ -2094,7 +2154,7 @@ definitions: - $ref: "#/definitions/base_dynamic_schema_loader" leads_stream: - $ref: "#/definitions/base_crm_search_incremental_stream" + $ref: "#/definitions/base_crm_search_incremental_deletion_stream" $parameters: name: leads entity: leads @@ -3296,6 +3356,22 @@ schemas: type: - "null" - boolean + archivedAt: + description: Date and time when the company was archived (deleted), if applicable + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false + _ab_cdc_deleted_at: + description: >- + CDC deletion marker. Non-null on tombstone records emitted for archived + (deleted) companies; equals archivedAt (fallback createdAt). + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false contacts: description: List of contacts associated with the company type: @@ -3365,6 +3441,22 @@ schemas: - "null" - object properties: + archivedAt: + description: Date and time when the record was archived (deleted), if applicable + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false + _ab_cdc_deleted_at: + description: >- + CDC deletion marker. Non-null on tombstone records emitted for archived + (deleted) records; equals archivedAt (fallback createdAt). + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false id: description: Unique identifier for the contact. type: @@ -3521,6 +3613,22 @@ schemas: - "null" - object properties: + archivedAt: + description: Date and time when the record was archived (deleted), if applicable + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false + _ab_cdc_deleted_at: + description: >- + CDC deletion marker. Non-null on tombstone records emitted for archived + (deleted) records; equals archivedAt (fallback createdAt). + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false id: description: Unique identifier for the deal type: @@ -4552,6 +4660,22 @@ schemas: - "null" - object properties: + archivedAt: + description: Date and time when the record was archived (deleted), if applicable + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false + _ab_cdc_deleted_at: + description: >- + CDC deletion marker. Non-null on tombstone records emitted for archived + (deleted) records; equals archivedAt (fallback createdAt). + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false id: description: Unique identifier for the call engagement. type: @@ -4619,6 +4743,22 @@ schemas: - "null" - object properties: + archivedAt: + description: Date and time when the record was archived (deleted), if applicable + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false + _ab_cdc_deleted_at: + description: >- + CDC deletion marker. Non-null on tombstone records emitted for archived + (deleted) records; equals archivedAt (fallback createdAt). + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false id: description: Unique identifier for the engagement email type: @@ -4753,6 +4893,22 @@ schemas: - "null" - object properties: + archivedAt: + description: Date and time when the record was archived (deleted), if applicable + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false + _ab_cdc_deleted_at: + description: >- + CDC deletion marker. Non-null on tombstone records emitted for archived + (deleted) records; equals archivedAt (fallback createdAt). + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false id: description: Unique identifier for the engagement note type: @@ -4820,6 +4976,22 @@ schemas: - "null" - object properties: + archivedAt: + description: Date and time when the record was archived (deleted), if applicable + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false + _ab_cdc_deleted_at: + description: >- + CDC deletion marker. Non-null on tombstone records emitted for archived + (deleted) records; equals archivedAt (fallback createdAt). + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false id: description: Unique identifier for the task type: @@ -5249,6 +5421,22 @@ schemas: - "null" - object properties: + archivedAt: + description: Date and time when the record was archived (deleted), if applicable + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false + _ab_cdc_deleted_at: + description: >- + CDC deletion marker. Non-null on tombstone records emitted for archived + (deleted) records; equals archivedAt (fallback createdAt). + type: + - "null" + - string + format: date-time + __ab_apply_cast_datetime: false id: description: Unique identifier for the lead type: diff --git a/airbyte-integrations/connectors/source-hubspot/metadata.yaml b/airbyte-integrations/connectors/source-hubspot/metadata.yaml index 1cc9f580f9d3..ab29de660107 100644 --- a/airbyte-integrations/connectors/source-hubspot/metadata.yaml +++ b/airbyte-integrations/connectors/source-hubspot/metadata.yaml @@ -10,7 +10,7 @@ data: connectorSubtype: api connectorType: source definitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c - dockerImageTag: 6.7.0 + dockerImageTag: 6.8.0 dockerRepository: airbyte/source-hubspot documentationUrl: https://docs.airbyte.com/integrations/sources/hubspot resourceRequirements: diff --git a/airbyte-integrations/connectors/source-hubspot/unit_tests/test_hubspot_deletion_retriever.py b/airbyte-integrations/connectors/source-hubspot/unit_tests/test_hubspot_deletion_retriever.py new file mode 100644 index 000000000000..83ee53364211 --- /dev/null +++ b/airbyte-integrations/connectors/source-hubspot/unit_tests/test_hubspot_deletion_retriever.py @@ -0,0 +1,64 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# +"""Unit test for HubspotDeletionRetriever: an archived record becomes a tombstone.""" + +components_module = __import__("components") +HubspotDeletionRetriever = components_module.HubspotDeletionRetriever + +ARCHIVED_AT = "2024-05-01T10:00:00.000Z" + +# Plain string schema so normalization is a no-op and the archivedAt string is preserved verbatim. +RECORDS_SCHEMA = { + "type": "object", + "additionalProperties": True, + "properties": { + "id": {"type": ["null", "string"]}, + "updatedAt": {"type": ["null", "string"]}, + "archivedAt": {"type": ["null", "string"]}, + "_ab_cdc_deleted_at": {"type": ["null", "string"]}, + "properties_name": {"type": ["null", "string"]}, + }, +} + + +class _FakeArchivedRetriever: + def __init__(self, records): + self._records = records + + def read_records(self, records_schema, stream_slice=None): + for record in self._records: + yield dict(record) + + +class _NoopLiveRetriever: + def read_records(self, records_schema, stream_slice=None): + return iter(()) + + +def test_archived_record_becomes_tombstone(): + archived = { + "id": "99", + "archived": True, + "createdAt": "2023-01-01T00:00:00.000Z", + "archivedAt": ARCHIVED_AT, + "properties": {"name": "Deleted Co"}, + } + retriever = HubspotDeletionRetriever( + live_retriever=_NoopLiveRetriever(), + config={}, + parameters={}, + archived_retriever=_FakeArchivedRetriever([archived]), + ) + + tombstones = list(retriever._read_archived_records(RECORDS_SCHEMA)) + + assert len(tombstones) == 1 + tombstone = tombstones[0] + assert tombstone["id"] == "99" + # archivedAt drives both the deletion marker and the dedup cursor. + assert tombstone["_ab_cdc_deleted_at"] == ARCHIVED_AT + assert tombstone["updatedAt"] == ARCHIVED_AT + assert tombstone["archivedAt"] == ARCHIVED_AT + # Full flattened property columns are present (not a sparse record). + assert tombstone["properties_name"] == "Deleted Co" diff --git a/docs/integrations/sources/hubspot.md b/docs/integrations/sources/hubspot.md index 13c9fb1e3143..46c7469f323d 100644 --- a/docs/integrations/sources/hubspot.md +++ b/docs/integrations/sources/hubspot.md @@ -431,6 +431,7 @@ If you use Airbyte Cloud and your organization restricts access to specific IPs, | Version | Date | Pull Request | Subject | |:------------|:-----------|:---------------------------------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| 6.8.0 | 2026-07-01 | [PR-XXXXX](https://github.com/airbytehq/airbyte/pull/XXXXX) | Capture CRM deletions: emit archived (deleted) objects as in-stream `_ab_cdc_deleted_at` tombstones for companies, contacts, deals, leads, and engagements (calls, emails, notes, tasks) | | 6.7.0 | 2026-06-11 | [76396](https://github.com/airbytehq/airbyte/pull/76396) | Add `treat_numbers_and_booleans_as_strings` config toggle to coerce dynamic `number`/`boolean` properties to `string` | | 6.6.1 | 2026-06-10 | [79636](https://github.com/airbytehq/airbyte/pull/79636) | Add configurable `property_history_lookback_window` (minutes) to property history streams (deals, contacts, companies) to prevent silent record loss caused by cursor drift from HubSpot calculated properties. Clarify existing `lookback_window` field as CRM Search-specific. | | 6.6.0 | 2026-06-08 | [71259](https://github.com/airbytehq/airbyte/pull/71259) | Add association streams for standard and custom objects, including optional OAuth scopes needed to support them |