🐛 Source Delighted: Fix pagination for survey_responses, bounces and unsubscribes streams (#9275)

* 9271 Fix pagination for delighted source

* 9271 Implement change request

* 9271 Implement change request

* 9271 Bump connector's version

Author: Zirochkaa

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

@@ -147,7 +147,7 @@
 - name: Delighted
   sourceDefinitionId: cc88c43f-6f53-4e8a-8c4d-b284baaf9635
   dockerRepository: airbyte/source-delighted
-  dockerImageTag: 0.1.0
+  dockerImageTag: 0.1.1
   documentationUrl: https://docs.airbyte.io/integrations/sources/delighted
   icon: delighted.svg
   sourceType: api

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

@@ -1280,7 +1280,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-delighted:0.1.0"
+- dockerImage: "airbyte/source-delighted:0.1.1"
   spec:
     documentationUrl: "https://docsurl.com"
     connectionSpecification:

airbyte-integrations/connectors/source-delighted/Dockerfile

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.version=0.1.1
 LABEL io.airbyte.name=airbyte/source-delighted

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

@@ -15,7 +15,7 @@ tests:
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
       empty_streams: ["bounces"]
-  incremental: # TODO if your connector does not implement incremental sync, remove this block
+  incremental:
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
       future_state_path: "integration_tests/abnormal_state.json"

airbyte-integrations/connectors/source-delighted/source_delighted/schemas/survey_responses.json

@@ -27,7 +27,21 @@
       "type": ["null", "integer"]
     },
     "person_properties": {
-      "type": "object"
+      "type": ["object", "null"],
+      "properties": {
+        "Delighted Source": {
+          "type": ["null", "string"]
+        },
+        "Delighted Device Type": {
+          "type": ["null", "string"]
+        },
+        "Delighted Operating System": {
+          "type": ["null", "string"]
+        },
+        "Delighted Browser": {
+          "type": ["null", "string"]
+        }
+      }
     },
     "notes": {
       "type": "array",
@@ -76,12 +90,10 @@
                     "text": {
                       "type": "string"
                     }
-                  },
-                  "required": ["id", "text"]
+                  }
                 }
               }
-            },
-            "required": ["free_response", "scale", "select_one", "select_many"]
+            }
           },
           "question": {
             "type": "object",
@@ -95,26 +107,10 @@
               "text": {
                 "type": "string"
               }
-            },
-            "required": ["id", "type", "text"]
+            }
           }
-        },
-        "required": ["id", "value", "question"]
+        }
       }
     }
-  },
-  "required": [
-    "id",
-    "person",
-    "survey_type",
-    "score",
-    "comment",
-    "permalink",
-    "created_at",
-    "updated_at",
-    "person_properties",
-    "notes",
-    "tags",
-    "additional_answers"
-  ]
+  }
 }

airbyte-integrations/connectors/source-delighted/source_delighted/source.py

@@ -23,6 +23,7 @@ class DelightedStream(HttpStream, ABC):
 
     # Page size
     limit = 100
+    page = 1
 
     # Define primary key to all streams as primary key
     primary_key = "id"
@@ -32,12 +33,10 @@ def __init__(self, since: int, **kwargs):
         self.since = since
 
     def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
-        # Getting next page link
-        next_page = response.links.get("next", None)
-        if next_page:
-            return dict(parse_qsl(urlparse(next_page.get("url")).query))
-        else:
-            return None
+        response_data = response.json()
+        if len(response_data) == self.limit:
+            self.page += 1
+            return {"page": self.page}
 
     def request_params(
         self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, any] = None, next_page_token: Mapping[str, Any] = None
@@ -49,8 +48,7 @@ def request_params(
         return params
 
     def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
-        records = response.json()
-        yield from records
+        yield from response.json()
 
 
 class IncrementalDelightedStream(DelightedStream, ABC):
@@ -77,38 +75,36 @@ def request_params(self, stream_state=None, **kwargs):
 
 
 class People(IncrementalDelightedStream):
-    def path(
-        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
-    ) -> str:
+    def path(self, **kwargs) -> str:
         return "people.json"
 
+    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
+        # Getting next page link
+        next_page = response.links.get("next", None)
+        if next_page:
+            return {"page_info": dict(parse_qsl(urlparse(next_page.get("url")).query)).get("page_info")}
+
 
 class Unsubscribes(IncrementalDelightedStream):
     cursor_field = "unsubscribed_at"
     primary_key = "person_id"
 
-    def path(
-        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
-    ) -> str:
+    def path(self, **kwargs) -> str:
         return "unsubscribes.json"
 
 
 class Bounces(IncrementalDelightedStream):
     cursor_field = "bounced_at"
     primary_key = "person_id"
 
-    def path(
-        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
-    ) -> str:
+    def path(self, **kwargs) -> str:
         return "bounces.json"
 
 
 class SurveyResponses(IncrementalDelightedStream):
     cursor_field = "updated_at"
 
-    def path(
-        self, stream_state: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
-    ) -> str:
+    def path(self, **kwargs) -> str:
         return "survey_responses.json"
 
     def request_params(self, stream_state=None, **kwargs):
@@ -148,4 +144,9 @@ def check_connection(self, logger, config) -> Tuple[bool, any]:
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         auth = self._get_authenticator(config)
         args = {"authenticator": auth, "since": config["since"]}
-        return [People(**args), Unsubscribes(**args), Bounces(**args), SurveyResponses(**args)]
+        return [
+            Bounces(**args),
+            People(**args),
+            SurveyResponses(**args),
+            Unsubscribes(**args),
+        ]

docs/integrations/sources/delighted.md

@@ -37,4 +37,5 @@ This connector supports `API PASSWORD` as the authentication method.
 
 | Version | Date | Pull Request | Subject |
 | :--- | :--- | :--- | :--- |
-| 0.1.0 | 2021-10-27 | [4551](https://github.com/airbytehq/airbyte/pull/4551) | Add Delighted source connector |
+| 0.1.1 | 2022-01-04 | [9275](https://github.com/airbytehq/airbyte/pull/9275) | Fix pagination handling for `survey_responses`, `bounces` and `unsubscribes` streams |
+| 0.1.0 | 2021-10-27 | [4551](https://github.com/airbytehq/airbyte/pull/4551) | Add Delighted source connector |