HGI-5916 / Support for custom objects

.vscode/.gitignore

@@ -0,0 +1,10 @@
+# IMPORTANT! This folder is hidden from git - if you need to store config files or other secrets,
+# make sure those are never staged for commit into your git repo. You can store them here or another
+# secure location.
+#
+# Note: This may be redundant with the global .gitignore for, and is provided
+# for redundancy. If the `.secrets` folder is not needed, you may delete it
+# from the project.
+
+*
+!.gitignore

tap_hubspot_beta/client_v3.py

@@ -224,6 +224,23 @@ def post_process(self, row: dict, context: Optional[dict]) -> dict:
         return row
 
 
+class DynamicDiscoveredHubspotV3Stream(hubspotV3Stream):
+    def post_process(self, row: dict, context: Optional[dict]) -> dict:
+        """As needed, append or transform raw data to match expected structure."""
+        for name, value in row["properties"].items():
+            row[name] = value
+        del row["properties"]
+        return row
+
+
+    def get_url_params(
+        self, context: Optional[dict], next_page_token: Optional[Any]
+    ) -> Dict[str, Any]:
+        """Return a dictionary of values to be used in URL parameterization."""
+        params = super().get_url_params(context, next_page_token)
+        params["properties"] = ",".join(self.selected_properties)
+        return params
+
 class hubspotV3SingleSearchStream(hubspotStream):
     """hubspot stream class."""
 

tap_hubspot_beta/streams.py

@@ -1717,3 +1717,84 @@ class AssociationTasksDealsStream(AssociationTasksStream):
 
     name = "associations_tasks_deals"
     path = "crm/v4/associations/tasks/deals/batch/read"
+
+
+class DiscoverCustomObjectsStream(hubspotV3Stream):
+    name = "discover_stream"
+    path = "crm-object-schemas/v3/schemas"
+    primary_keys = ["id"]
+    replication_key = "updatedAt"
+
+    schema = th.PropertiesList(
+        th.Property("labels", th.ObjectType(
+            th.Property("singular", th.StringType),
+            th.Property("plural", th.StringType)
+        )),
+        th.Property("requiredProperties", th.ArrayType(th.StringType)),
+        th.Property("searchableProperties", th.ArrayType(th.StringType)),
+        th.Property("primaryDisplayProperty", th.StringType),
+        th.Property("secondaryDisplayProperties", th.ArrayType(th.StringType)),
+        th.Property("description", th.StringType),
+        th.Property("archived", th.BooleanType),
+        th.Property("restorable", th.BooleanType),
+        th.Property("metaType", th.StringType),
+        th.Property("id", th.StringType),
+        th.Property("fullyQualifiedName", th.StringType),
+        th.Property("createdAt", th.DateTimeType),
+        th.Property("updatedAt", th.DateTimeType),
+        th.Property("createdByUserId", th.IntegerType),
+        th.Property("updatedByUserId", th.IntegerType),
+        th.Property("objectTypeId", th.StringType),
+        th.Property("properties", th.ArrayType(th.ObjectType(
+            th.Property("name", th.StringType),
+            th.Property("label", th.StringType),
+            th.Property("type", th.StringType),
+            th.Property("fieldType", th.StringType),
+            th.Property("description", th.StringType),
+            th.Property("groupName", th.StringType),
+            th.Property("options", th.ArrayType(th.ObjectType(
+                th.Property("label", th.StringType),
+                th.Property("value", th.StringType),
+                th.Property("description", th.StringType),
+                th.Property("displayOrder", th.IntegerType),
+                th.Property("hidden", th.BooleanType)
+            ))),
+            th.Property("createdUserId", th.StringType),
+            th.Property("updatedUserId", th.StringType),
+            th.Property("displayOrder", th.IntegerType),
+            th.Property("calculated", th.BooleanType),
+            th.Property("externalOptions", th.BooleanType),
+            th.Property("archived", th.BooleanType),
+            th.Property("hasUniqueValue", th.BooleanType),
+            th.Property("hidden", th.BooleanType),
+            th.Property("modificationMetadata", th.ObjectType(
+                th.Property("archivable", th.BooleanType),
+                th.Property("readOnlyDefinition", th.BooleanType),
+                th.Property("readOnlyValue", th.BooleanType)
+            )),
+            th.Property("formField", th.BooleanType),
+            th.Property("dataSensitivity", th.StringType),
+            th.Property("showCurrencySymbol", th.BooleanType),
+            th.Property("hubspotDefined", th.BooleanType),
+            th.Property("createdAt", th.DateTimeType),
+            th.Property("updatedAt", th.DateTimeType),
+            th.Property("archivedAt", th.DateTimeType),
+            th.Property("referencedObjectType", th.StringType),
+            th.Property("calculationFormula", th.StringType),
+        ))),
+        th.Property("associations", th.ArrayType(th.ObjectType(
+            th.Property("fromObjectTypeId", th.StringType),
+            th.Property("toObjectTypeId", th.StringType),
+            th.Property("name", th.StringType),
+            th.Property("cardinality", th.StringType),
+            th.Property("inverseCardinality", th.StringType),
+            th.Property("hasUserEnforcedMaxToObjectIds", th.BooleanType),
+            th.Property("hasUserEnforcedMaxFromObjectIds", th.BooleanType),
+            th.Property("maxToObjectIds", th.IntegerType),
+            th.Property("maxFromObjectIds", th.IntegerType),
+            th.Property("id", th.StringType),
+            th.Property("createdAt", th.StringType),
+            th.Property("updatedAt", th.StringType),
+        ))),
+        th.Property("name", th.StringType),
+    ).to_dict()

tap_hubspot_beta/tap.py

@@ -1,10 +1,12 @@
 """hubspot tap class."""
 
-from typing import List
+from typing import Any, Dict, List
 
 from singer_sdk import Stream, Tap
 from singer_sdk import typing as th
+from singer_sdk.exceptions import FatalAPIError
 
+from tap_hubspot_beta.client_v3 import hubspotV3Stream, DynamicDiscoveredHubspotV3Stream
 from tap_hubspot_beta.streams import (
     AccountStream,
     AssociationDealsCompaniesStream,
@@ -69,7 +71,8 @@
     AssociationTasksCompaniesStream,
     AssociationTasksContactsStream,
     AssociationTasksDealsStream,
-    CampaignsStream
+    DiscoverCustomObjectsStream,
+    CampaignsStream,
 )
 
 STREAM_TYPES = [
@@ -139,7 +142,6 @@
     CampaignsStream
 ]
 
-
 class Taphubspot(Tap):
     """hubspot tap class."""
 
@@ -168,7 +170,15 @@ def __init__(
 
     def discover_streams(self) -> List[Stream]:
         """Return a list of discovered streams."""
-        return [stream_class(tap=self) for stream_class in STREAM_TYPES]
+        streams = [stream_class(tap=self) for stream_class in STREAM_TYPES]
+        try:
+            discover_stream = DiscoverCustomObjectsStream(tap=self)
+            for record in discover_stream.get_records(context={}):
+                stream_class = self.generate_stream_class(record)
+                streams.append(stream_class(tap=self))
+        except FatalAPIError as exc:
+            self.logger.info(f"failed to discover custom objects. Error={exc}")
+        return streams
 
     @property
     def catalog_dict(self) -> dict:
@@ -189,6 +199,65 @@ def catalog_dict(self) -> dict:
                         stream["schema"]["properties"][field]["field_meta"] = stream_class.fields_metadata.get(field, {})
         return catalog
 
+    def generate_stream_class(self, custom_object: Dict[str, Any]) -> hubspotV3Stream:
+        # check for required fields to construct the custom objects class
+        required_fields = ["id", "name", "objectTypeId", "properties"]
+        errors = []
+        for field in required_fields:
+            if not custom_object.get(field):
+                errors.append(f"Missing {field} in custom object.")
+        if errors:
+            errors.append(f"Failed custom_object={custom_object}.")
+            error_msg = "\n".join(errors)
+            raise ValueError(error_msg)
+
+        name = custom_object.get("name")
+        object_type_id = custom_object.get("objectTypeId")
+        properties = custom_object.get("properties")
+
+        if custom_object.get("archived", False):
+            name = "archived_" + name
+        class_name = name + "_Stream"
+        class_name = "".join(word.capitalize() for word in class_name.split("_"))
+
+        self.logger.info(f"Creating class {class_name}")
+
+        return type(
+            class_name,
+            (DynamicDiscoveredHubspotV3Stream,),
+            {
+                "name": name,
+                "path": f"crm/v3/objects/{object_type_id}/",
+                "records_jsonpath": "$.results[*]",
+                "primary_keys": ["id"],
+                "replication_key": "updatedAt",
+                "page_size": 100,
+                "schema": self.generate_schema(properties),
+                "is_custom_stream": True,
+            },
+        )
+
+    def generate_schema(self, properties: List[Dict[str, Any]]) -> dict:
+        properties_list = [
+            th.Property("id", th.StringType),
+            th.Property("updatedAt", th.DateTimeType), 
+            th.Property("createdAt", th.DateTimeType), 
+            th.Property("archived", th.BooleanType)
+        ]
+        main_properties = [p.name for p in properties_list]
+
+        for property in properties:
+            field_name = property.get("name")
+            if field_name in main_properties:
+                self.logger.info(f"Skipping field, it is a default field and already included.")
+                continue
+            if not field_name:
+                self.logger.info(f"Skipping field without name.")
+                continue
+            th_type = hubspotV3Stream.extract_type(property, self.config.get("type_booleancheckbox_as_boolean"))
+            properties_list.append(th.Property(field_name, th_type))
+        return th.PropertiesList(*properties_list).to_dict()
+
 
 if __name__ == "__main__":
     Taphubspot.cli()