move CRM endpoints to search first

Author: kevin-induro

sources/hubspot/__init__.py

@@ -30,28 +30,32 @@
     List,
     Literal,
     Optional,
-    Sequence,
 )
 from urllib.parse import quote
 
 import dlt
-from dlt.common import pendulum
+from dlt.common import logger, pendulum
 from dlt.common.typing import TDataItems
 from dlt.sources import DltResource
 
 from .helpers import (
     _get_property_names_types,
     _to_dlt_columns_schema,
+    search_data,
     fetch_data,
     fetch_property_history,
     get_properties_labels,
+    SearchOutOfBoundsException,
 )
 from .settings import (
     ALL_OBJECTS,
     ARCHIVED_PARAM,
     CRM_OBJECT_ENDPOINTS,
     CRM_PIPELINES_ENDPOINT,
+    CRM_SEARCH_OBJECT_ENDPOINTS,
     ENTITY_PROPERTIES,
+    LAST_MODIFIED_PROPERTY,
+    HUBSPOT_CREATION_DATE,
     MAX_PROPS_LENGTH,
     OBJECT_TYPE_PLURAL,
     OBJECT_TYPE_SINGULAR,
@@ -73,6 +77,7 @@ def fetch_data_for_properties(
     api_key: str,
     object_type: str,
     soft_delete: bool,
+    last_modified: str = None,
 ) -> Iterator[TDataItems]:
     """
     Fetch data for a given set of properties from the HubSpot API.
@@ -82,20 +87,56 @@ def fetch_data_for_properties(
         api_key (str): HubSpot API key for authentication.
         object_type (str): The type of HubSpot object (e.g., 'company', 'contact').
         soft_delete (bool): Flag to fetch soft-deleted (archived) records.
+        last_modified (str): The date from which to fetch records. If None, get all records.
 
     Yields:
         Iterator[TDataItems]: Data retrieved from the HubSpot API.
     """
+    logger.info(f"Fetching data for {object_type}.")
     # The Hubspot API expects a comma separated string as properties
     joined_props = ",".join(sorted(props))
     params: Dict[str, Any] = {"properties": joined_props, "limit": 100}
     context: Optional[Dict[str, Any]] = (
         {SOFT_DELETE_KEY: False} if soft_delete else None
     )
 
-    yield from fetch_data(
-        CRM_OBJECT_ENDPOINTS[object_type], api_key, params=params, context=context
-    )
+    if last_modified is not None:
+        logger.info(f"Attempting search starting at {last_modified}.")
+        search_params: Dict[str, Any] = {
+            "properties": sorted(props),
+            "limit": 200,
+            "filterGroups": [
+                {
+                    "filters": [
+                        {
+                            "propertyName": LAST_MODIFIED_PROPERTY[object_type],
+                            "operator": "GTE",
+                            "value": last_modified,
+                        }
+                    ]
+                }
+            ],
+        }
+
+        try:
+            yield from search_data(
+                CRM_SEARCH_OBJECT_ENDPOINTS[object_type],
+                api_key,
+                params=search_params,
+                context=context,
+            )
+        except SearchOutOfBoundsException:
+            logger.info("Search out of bounds, fetching all data")
+            yield from fetch_data(
+                CRM_OBJECT_ENDPOINTS[object_type],
+                api_key,
+                params=params,
+                context=context,
+            )
+    else:
+        yield from fetch_data(
+            CRM_OBJECT_ENDPOINTS[object_type], api_key, params=params, context=context
+        )
     if soft_delete:
         yield from fetch_data(
             CRM_OBJECT_ENDPOINTS[object_type],
@@ -109,6 +150,7 @@ def crm_objects(
     object_type: str,
     api_key: str,
     props: List[str],
+    last_modified: dlt.sources.incremental[str],
     include_custom_props: bool = True,
     archived: bool = False,
 ) -> Iterator[TDataItems]:
@@ -119,6 +161,7 @@ def crm_objects(
         object_type (str): Type of HubSpot object (e.g., 'company', 'contact').
         api_key (str): API key for HubSpot authentication.
         props (List[str]): List of properties to retrieve.
+        last_modified (str): The date from which to fetch records
         include_custom_props (bool, optional): Include custom properties in the result. Defaults to True.
         archived (bool, optional): Fetch archived (soft-deleted) objects. Defaults to False.
 
@@ -135,8 +178,17 @@ def crm_objects(
         prop: _to_dlt_columns_schema({prop: hb_type})
         for prop, hb_type in props_to_type.items()
     }
+    last_modified_on = (
+        None
+        if last_modified.start_value == last_modified.initial_value
+        else last_modified.start_value
+    )
     for batch in fetch_data_for_properties(
-        list(props_to_type.keys()), api_key, object_type, archived
+        list(props_to_type.keys()),
+        api_key,
+        object_type,
+        archived,
+        last_modified_on,
     ):
         yield dlt.mark.with_hints(batch, dlt.mark.make_hints(columns=col_type_hints))
 
@@ -176,7 +228,7 @@ def crm_object_history(
     # This is especially relevant for columns of type "number" in Hubspot
     # that are returned as strings by the API
     for batch in fetch_property_history(
-        CRM_OBJECT_ENDPOINTS[object_type],
+        CRM_SEARCH_OBJECT_ENDPOINTS[object_type],
         api_key,
         ",".join(sorted(props_to_type.keys())),
     ):
@@ -411,6 +463,10 @@ def get_pipelines(object_type: str) -> Iterator[TDataItems]:
             object_type=obj,
             api_key=api_key,
             props=properties.get(obj),
+            last_modified=dlt.sources.incremental(
+                LAST_MODIFIED_PROPERTY[obj],
+                initial_value=HUBSPOT_CREATION_DATE.isoformat(),
+            ),
             include_custom_props=include_custom_props,
             archived=soft_delete,
         )

sources/hubspot/helpers.py

@@ -13,6 +13,10 @@
 BASE_URL = "https://api.hubapi.com/"
 
 
+class SearchOutOfBoundsException(Exception):
+    pass
+
+
 def get_url(endpoint: str) -> str:
     """Get absolute hubspot endpoint URL"""
     return urllib.parse.urljoin(BASE_URL, endpoint)
@@ -48,6 +52,22 @@ def pagination(
         return None
 
 
+def search_pagination(
+    url: str,
+    _data: Dict[str, Any],
+    headers: Dict[str, Any],
+    params: Optional[Dict[str, Any]] = None,
+) -> Optional[Dict[str, Any]]:
+    _next = _data.get("paging", {}).get("next", False)
+    if _next:
+        after = _next["after"]
+        # Get the next page response
+        r = requests.post(url, headers=headers, json={**params, after: after})
+        return r.json()  # type: ignore
+    else:
+        return None
+
+
 def extract_association_data(
     _obj: Dict[str, Any],
     data: Dict[str, Any],
@@ -126,6 +146,34 @@ def fetch_property_history(
             _data = None
 
 
+def search_data(
+    endpoint: str,
+    api_key: str,
+    params: Optional[Dict[str, Any]] = None,
+    context: Optional[Dict[str, Any]] = None,
+) -> Iterator[List[Dict[str, Any]]]:
+    # Construct the URL and headers for the API request
+    url = get_url(endpoint)
+    headers = _get_headers(api_key)
+
+    # Make the API request
+    r = requests.post(url, headers=headers, json=params)
+    # Parse the API response and yield the properties of each result
+    # Parse the response JSON data
+    _data = r.json()
+
+    if _data.get("total", 0) > 9999:
+        raise SearchOutOfBoundsException
+    else:
+        # Yield the properties of each result in the API response
+        while _data is not None:
+            if "results" in _data:
+                yield _data_to_objects(_data, headers, context)
+
+            # Follow pagination links if they exist
+            _data = search_pagination(url, _data, headers, params)
+
+
 def fetch_data(
     endpoint: str,
     api_key: str,
@@ -168,38 +216,43 @@ def fetch_data(
     # Parse the API response and yield the properties of each result
     # Parse the response JSON data
     _data = r.json()
+
     # Yield the properties of each result in the API response
     while _data is not None:
         if "results" in _data:
-            _objects: List[Dict[str, Any]] = []
-            for _result in _data["results"]:
-                _obj = _result.get("properties", _result)
-                if "id" not in _obj and "id" in _result:
-                    # Move id from properties to top level
-                    _obj["id"] = _result["id"]
-                if "associations" in _result:
-                    for association in _result["associations"]:
-                        __data = _result["associations"][association]
-
-                        __values = extract_association_data(
-                            _obj, __data, association, headers
-                        )
-
-                        # remove duplicates from list of dicts
-                        __values = [
-                            dict(t) for t in {tuple(d.items()) for d in __values}
-                        ]
-
-                        _obj[association] = __values
-                if context:
-                    _obj.update(context)
-                _objects.append(_obj)
-            yield _objects
+            yield _data_to_objects(_data, headers, context)
 
         # Follow pagination links if they exist
         _data = pagination(_data, headers)
 
 
+def _data_to_objects(
+    data: Any,
+    headers: Dict[str, str],
+    context: Optional[Dict[str, Any]] = None,
+) -> List[Dict[str, Any]]:
+    _objects: List[Dict[str, Any]] = []
+    for _result in data["results"]:
+        _obj = _result.get("properties", _result)
+        if "id" not in _obj and "id" in _result:
+            # Move id from properties to top level
+            _obj["id"] = _result["id"]
+        if "associations" in _result:
+            for association in _result["associations"]:
+                __data = _result["associations"][association]
+
+                __values = extract_association_data(_obj, __data, association, headers)
+
+                # remove duplicates from list of dicts
+                __values = [dict(t) for t in {tuple(d.items()) for d in __values}]
+
+                _obj[association] = __values
+        if context:
+            _obj.update(context)
+        _objects.append(_obj)
+    return _objects
+
+
 def _get_property_names_types(
     api_key: str, object_type: str
 ) -> Dict[str, Union[str, None]]:

sources/hubspot/settings.py

@@ -3,8 +3,25 @@
 from dlt.common import pendulum
 from dlt.common.data_types import TDataType
 
+HUBSPOT_CREATION_DATE = pendulum.datetime(year=2006, month=6, day=1)
 STARTDATE = pendulum.datetime(year=2024, month=2, day=10)
 
+CRM_CONTACTS_SEARCH_ENDPOINT = "/crm/v3/objects/contacts/search"
+CRM_COMPANIES_SEARCH_ENDPOINT = "/crm/v3/objects/companies/search"
+CRM_DEALS_SEARCH_ENDPOINT = "/crm/v3/objects/deals/search"
+CRM_PRODUCTS_SEARCH_ENDPOINT = "/crm/v3/objects/products/search"
+CRM_TICKETS_SEARCH_ENDPOINT = "/crm/v3/objects/tickets/search"
+CRM_QUOTES_SEARCH_ENDPOINT = "/crm/v3/objects/quotes/search"
+
+CRM_SEARCH_OBJECT_ENDPOINTS = {
+    "contact": CRM_CONTACTS_SEARCH_ENDPOINT,
+    "company": CRM_COMPANIES_SEARCH_ENDPOINT,
+    "deal": CRM_DEALS_SEARCH_ENDPOINT,
+    "product": CRM_PRODUCTS_SEARCH_ENDPOINT,
+    "ticket": CRM_TICKETS_SEARCH_ENDPOINT,
+    "quote": CRM_QUOTES_SEARCH_ENDPOINT,
+}
+
 CRM_CONTACTS_ENDPOINT = (
     "/crm/v3/objects/contacts?associations=deals,products,tickets,quotes"
 )
@@ -112,6 +129,14 @@
     "quote": DEFAULT_QUOTE_PROPS,
 }
 
+LAST_MODIFIED_PROPERTY = {
+    "company": "hs_lastmodifieddate",
+    "contact": "lastmodifieddate",
+    "deal": "hs_lastmodifieddate",
+    "ticket": "hs_lastmodifieddate",
+    "product": "hs_lastmodifieddate",
+    "quote": "hs_lastmodifieddate",
+}
 
 PIPELINES_OBJECTS = ["deals", "tickets"]
 SOFT_DELETE_KEY = "is_deleted"