Source Hubspot: Fix getting submissions for all forms (#9555)

Co-authored-by: Marcos Marx <marcosmarxm@gmail.com>

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.36",
+  "dockerImageTag": "0.1.37",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/hubspot",
   "icon": "hubspot.svg"
 }

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

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

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

@@ -3082,7 +3082,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-hubspot:0.1.36"
+- dockerImage: "airbyte/source-hubspot:0.1.37"
   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.36
+LABEL io.airbyte.version=0.1.37
 LABEL io.airbyte.name=airbyte/source-hubspot

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

@@ -348,10 +348,14 @@ def _read(self, getter: Callable, params: MutableMapping[str, Any] = None) -> It
             if not next_page_token:
                 break
 
-    def read(self, getter: Callable, params: Mapping[str, Any] = None) -> Iterator:
+    def read(self, getter: Callable, params: Mapping[str, Any] = None, filter_old_records: bool = True) -> Iterator:
         default_params = {self.limit_field: self.limit}
         params = {**default_params, **params} if params else {**default_params}
-        yield from self._filter_old_records(self._read(getter, params))
+        generator = self._read(getter, params)
+        if filter_old_records:
+            generator = self._filter_old_records(generator)
+
+        yield from generator
 
     def parse_response(self, response: Union[Mapping[str, Any], List[dict]]) -> Iterator:
         if isinstance(response, Mapping):
@@ -879,6 +883,10 @@ class FormSubmissionStream(Stream):
     limit = 50
     updated_at_field = "updatedAt"
 
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.forms = FormStream(**kwargs)
+
     def _transform(self, records: Iterable) -> Iterable:
         for record in super()._transform(records):
             keys = record.keys()
@@ -891,10 +899,14 @@ def _transform(self, records: Iterable) -> Iterable:
             yield record
 
     def list_records(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
+        seen = set()
+        # To get submissions for all forms date filtering has to be disabled
+        for form in self.forms.read(getter=partial(self.forms._api.get, url=self.forms.url), filter_old_records=False):
+            if form["id"] not in seen:
+                seen.add(form["id"])
+                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):