Source Hubspot: adding form_submissions stream (#8011)

* Source Hubspot: adding form_submissions stream

* Adding form_id to form_submissions output

* Adding form_id to form_submissions output and schema

* Removing field deletions and id generation

* Adding new line to form_submissions.json

* Fixing form_submissions.json schema

* add form_submissions as empty_stream to run ci airbyte

* add forms_submissions to empty_stream

* bump connector seed version

Co-authored-by: Marcos Marx <[email protected]>

Author: gergelylendvai

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.30",
+  "dockerImageTag": "0.1.32",
   "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.31
+  dockerImageTag: 0.1.32
   documentationUrl: https://docs.airbyte.io/integrations/sources/hubspot
   icon: hubspot.svg
   sourceType: api

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

@@ -2846,7 +2846,7 @@
               path_in_connector_config:
               - "credentials"
               - "client_secret"
-- dockerImage: "airbyte/source-hubspot:0.1.31"
+- dockerImage: "airbyte/source-hubspot:0.1.32"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/hubspot"
     connectionSpecification:

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.31
+LABEL io.airbyte.version=0.1.32
 LABEL io.airbyte.name=airbyte/source-hubspot

airbyte-integrations/connectors/source-hubspot/acceptance-test-config.yml

@@ -18,14 +18,14 @@ tests:
   basic_read:
     - config_path: "secrets/config.json"
       configured_catalog_path: "sample_files/full_refresh_catalog.json"
-      empty_streams: ["workflows"]
+      empty_streams: ["workflows", "form_submissions"]
     - config_path: "secrets/config_oauth.json"
       configured_catalog_path: "sample_files/configured_catalog_for_oauth_config.json"
       # The `campaigns` stream is empty in this case, because we use a catalog with
       # incremental streams: subscription_changes and email_events (it takes a long time to read)
       # and therefore the start date is set at 2021-10-10 for `config_oauth.json`,
       # but the campaign was created on 2021-01-11
-      empty_streams: ["campaigns", "workflows", "contacts_list_memberships"]
+      empty_streams: ["campaigns", "workflows", "contacts_list_memberships", "form_submissions"]
   incremental:
     - config_path: "secrets/config.json"
       configured_catalog_path: "sample_files/configured_catalog.json"

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

@@ -93,6 +93,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

@@ -81,6 +81,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

@@ -732,6 +732,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

@@ -19,6 +19,7 @@
     EmailEventStream,
     EngagementStream,
     FormStream,
+    FormSubmissionStream,
     MarketingEmailStream,
     OwnerStream,
     SubscriptionChangeStream,
@@ -45,6 +46,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),