Source Hubspot: adding form_submissions stream

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/36c891d9-4bd9-43ac-bad2-10e12756272c.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "36c891d9-4bd9-43ac-bad2-10e12756272c",
   "name": "HubSpot",
   "dockerRepository": "airbyte/source-hubspot",
-  "dockerImageTag": "0.1.29",
+  "dockerImageTag": "0.1.30",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/hubspot",
   "icon": "hubspot.svg"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -287,7 +287,7 @@
 - name: HubSpot
   sourceDefinitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c
   dockerRepository: airbyte/source-hubspot
-  dockerImageTag: 0.1.29
+  dockerImageTag: 0.1.30
   documentationUrl: https://docs.airbyte.io/integrations/sources/hubspot
   icon: hubspot.svg
   sourceType: api

airbyte-integrations/connectors/source-hubspot/Dockerfile

@@ -34,5 +34,5 @@ COPY source_hubspot ./source_hubspot
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.29
+LABEL io.airbyte.version=0.1.30
 LABEL io.airbyte.name=airbyte/source-hubspot

airbyte-integrations/connectors/source-hubspot/sample_files/configured_catalog.json

@@ -96,6 +96,15 @@
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite"
     },
+    {
+      "stream": {
+        "name": "form_submissions",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    },
     {
       "stream": {
         "name": "line_items",

airbyte-integrations/connectors/source-hubspot/sample_files/configured_catalog_for_oauth_config.json

@@ -84,6 +84,15 @@
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite"
     },
+    {
+      "stream": {
+        "name": "form_submissions",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    },
     {
       "stream": {
         "name": "line_items",

airbyte-integrations/connectors/source-hubspot/sample_files/full_refresh_catalog.json

@@ -72,6 +72,15 @@
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite"
     },
+    {
+      "stream": {
+        "name": "form_submissions",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    },
     {
       "stream": {
         "name": "line_items",

airbyte-integrations/connectors/source-hubspot/source_hubspot/api.py

@@ -702,6 +702,34 @@ class FormStream(Stream):
     created_at_field = "createdAt"
 
 
+class FormSubmissionStream(Stream):
+    """Marketing Forms, API v1
+    This endpoint requires the forms scope.
+    Docs: https://legacydocs.hubspot.com/docs/methods/forms/get-submissions-for-a-form
+    """
+
+    url = "/form-integrations/v1/submissions/forms"
+    limit = 50
+    updated_at_field = "updatedAt"
+
+    def _transform(self, records: Iterable) -> Iterable:
+        for record in super()._transform(records):
+            keys = record.keys()
+
+            # There's no updatedAt field in the submission however forms fetched by using this field,
+            # so it has to be added to the submissions otherwise it would fail when calling _filter_old_records
+            if "updatedAt" not in keys:
+                record["updatedAt"] = record["submittedAt"]
+
+            yield record
+
+    def list(self, fields) -> Iterable:
+        for form in self.read(getter=partial(self._api.get, url="/marketing/v3/forms")):
+            for submission in self.read(getter=partial(self._api.get, url=f"{self.url}/{form['id']}")):
+                submission["formId"] = form["id"]
+                yield submission
+
+
 class MarketingEmailStream(Stream):
     """Marketing Email, API v1
     Docs: https://legacydocs.hubspot.com/docs/methods/cms_email/get-all-marketing-emails

airbyte-integrations/connectors/source-hubspot/source_hubspot/client.py

@@ -18,6 +18,7 @@
     EmailEventStream,
     EngagementStream,
     FormStream,
+    FormSubmissionStream,
     MarketingEmailStream,
     OwnerStream,
     SubscriptionChangeStream,
@@ -43,6 +44,7 @@ def __init__(self, start_date, credentials, **kwargs):
             "email_events": EmailEventStream(**common_params),
             "engagements": EngagementStream(**common_params),
             "forms": FormStream(**common_params),
+            "form_submissions": FormSubmissionStream(**common_params),
             "line_items": CRMObjectIncrementalStream(entity="line_item", **common_params),
             "marketing_emails": MarketingEmailStream(**common_params),
             "owners": OwnerStream(**common_params),

airbyte-integrations/connectors/source-hubspot/source_hubspot/schemas/form_submissions.json

@@ -0,0 +1,35 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema",
+  "type": ["null", "object"],
+  "properties": {
+    "submittedAt": {
+      "type": ["null", "integer"]
+    },
+    "updatedAt": {
+      "type": ["null", "integer"]
+    },
+    "values": {
+      "type": ["null", "array"],
+      "items": {
+        "type": "object",
+        "properties": {
+          "name": {
+            "type": ["null", "string"]
+          },
+          "value": {
+            "type": ["null", "string"]
+          },
+          "objectTypeId": {
+            "type": ["null", "string"]
+          }
+        }
+      }
+    },
+    "pageUrl": {
+      "type": ["null", "string"]
+    },
+    "formId": {
+      "type": ["null", "string"]
+    }
+  }
+}