🎉Source Zenloop: migrate to lowcode

airbyte-integrations/connectors/source-zenloop/Dockerfile

@@ -34,5 +34,5 @@ COPY source_zenloop ./source_zenloop
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.3
+LABEL io.airbyte.version=0.1.4
 LABEL io.airbyte.name=airbyte/source-zenloop

airbyte-integrations/connectors/source-zenloop/README.md

@@ -82,7 +82,7 @@ docker run --rm -v $(pwd)/secrets:/secrets -v $(pwd)/integration_tests:/integrat
 Make sure to familiarize yourself with [pytest test discovery](https://docs.pytest.org/en/latest/goodpractices.html#test-discovery) to know how your test files and methods should be named.
 First install test dependencies into your virtual environment:
 ```
-pip install .[tests]
+pip install .'[tests]'
 ```
 ### Unit Tests
 To run unit tests locally, from the connector directory run:

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

@@ -13,31 +13,12 @@ tests:
     - config_path: "secrets/config.json"
   basic_read:
     - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_answers.json"
-      empty_streams: []
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_surveys.json"
-      empty_streams: []
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_answers_survey_group.json"
-      empty_streams: []
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_survey_groups.json"
-      empty_streams: []
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_properties.json"
+      configured_catalog_path: "integration_tests/configured_catalog.json"
       empty_streams: []
   incremental:
     - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_answers.json"
-      future_state_path: "integration_tests/abnormal_state.json"
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_answers_survey_group.json"
+      configured_catalog_path: "integration_tests/configured_catalog.json"
       future_state_path: "integration_tests/abnormal_state.json"
   full_refresh:
     - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_survey_groups.json"
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_surveys.json"
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog_properties.json"
+      configured_catalog_path: "integration_tests/configured_catalog.json"

airbyte-integrations/connectors/source-zenloop/integration_tests/configured_catalog.json

@@ -0,0 +1,49 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "answers",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"]
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "surveys",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    },
+    {
+      "stream": {
+        "name": "survey_groups",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    },
+    {
+      "stream": {
+        "name": "answers_survey_group",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"]
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "properties",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
+    }
+  ]
+}

airbyte-integrations/connectors/source-zenloop/source_zenloop/components.py

@@ -0,0 +1,53 @@
+#
+# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+#
+
+
+import math
+from dataclasses import dataclass
+from typing import Any, Iterable, List, Mapping, Optional
+
+import requests
+from airbyte_cdk.models import SyncMode
+from airbyte_cdk.sources.declarative.requesters.paginators.strategies.page_increment import PageIncrement
+from airbyte_cdk.sources.declarative.stream_slicers.datetime_stream_slicer import DatetimeStreamSlicer
+from airbyte_cdk.sources.declarative.stream_slicers.substream_slicer import SubstreamSlicer
+from airbyte_cdk.sources.declarative.types import StreamSlice, StreamState
+
+
+@dataclass
+class ZenloopDatetimeStreamSlicer(DatetimeStreamSlicer):
+    def get_request_params(
+        self,
+        *,
+        stream_state: Optional[StreamState] = None,
+        stream_slice: Optional[StreamSlice] = None,
+        next_page_token: Optional[Mapping[str, Any]] = None,
+    ) -> Mapping[str, Any]:
+        params = super().get_request_params(stream_state=stream_state, stream_slice=stream_slice, next_page_token=next_page_token)
+        date_shortcut = {"date_shortcut": "custom"}  # required request param when specified date_from and date_to params
+        params.update(date_shortcut)
+        return params
+
+
+@dataclass
+class ZenloopSubstreamSlicer(SubstreamSlicer):
+    def stream_slices(self, sync_mode: SyncMode, stream_state: StreamState) -> Iterable[StreamSlice]:
+        """
+
+        config_parent_field : parent field name in config
+
+        Use parent id's as stream state value if it specified in config or
+        create stream_slices according SubstreamSlicer logic.
+
+        """
+        config = self._options.get("config")
+        parent_field = self._options.get("config_parent_field")
+        custom_stream_state_value = config.get(parent_field)
+
+        if not custom_stream_state_value:
+            yield from super().stream_slices(sync_mode, stream_state)
+        else:
+            for parent_stream_config in self.parent_stream_configs:
+                stream_state_field = parent_stream_config.stream_slice_field or None
+                yield {stream_state_field: custom_stream_state_value, "parent_slice": {}}