Source Pipedrive: Add incremental sync to organizations stream

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

@@ -30,6 +30,9 @@ acceptance_tests:
           - name: stages
           - name: deal_products
           - name: mail
+          - name: deals
+          - name: users
+          - name: persons
       - config_path: "secrets/config.json"
         expect_records:
           path: "integration_tests/expected_records.jsonl"
@@ -52,7 +55,33 @@ acceptance_tests:
           - name: persons
 
   incremental:
-    bypass_reason: "All incremental streams are empty in sandbox account."
+    tests:
+      - config_path: secrets/config.json
+        configured_catalog_path: integration_tests/configured_catalog.json
+        future_state:
+          future_state_path: integration_tests/abnormal_state.json
+          missing_streams:
+            - name: deals
+              bypass_reason: "Stream is empty in sandbox account."
+            - name: files
+              bypass_reason: "Stream is empty in sandbox account."
+            - name: filters
+              bypass_reason: "Stream is empty in sandbox account."
+            - name: notes
+              bypass_reason: "Stream is empty in sandbox account."
+            - name: activities
+              bypass_reason: "Stream is empty in sandbox account."
+            - name: persons
+              bypass_reason: "Stream is empty in sandbox account."
+            - name: pipelines
+              bypass_reason: "Stream is empty in sandbox account."
+            - name: products
+              bypass_reason: "Stream is empty in sandbox account."
+            - name: stages
+              bypass_reason: "Stream is empty in sandbox account."
+            - name: users
+              bypass_reason: "Stream is empty in sandbox account."
+        skip_comprehensive_incremental_tests: true
   full_refresh:
     tests:
       - config_path: "secrets/config.json"

airbyte-integrations/connectors/source-pipedrive/components.py

@@ -1,7 +1,7 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-from dataclasses import InitVar, dataclass
+from dataclasses import InitVar, dataclass, field
 from typing import Any, List, Mapping, Union
 
 import requests
@@ -41,7 +41,7 @@ class NullCheckedDpathExtractor(RecordExtractor):
     nullable_nested_field: Union[InterpolatedString, str]
     config: Config
     parameters: InitVar[Mapping[str, Any]]
-    decoder: Decoder = JsonDecoder(parameters={})
+    decoder: Decoder = field(default_factory=lambda: JsonDecoder(parameters={}))
 
     def __post_init__(self, parameters: Mapping[str, Any]):
         self._dpath_extractor = DpathExtractor(

airbyte-integrations/connectors/source-pipedrive/integration_tests/abnormal_state.json

@@ -124,7 +124,7 @@
     "type": "STREAM",
     "stream": {
       "stream_state": {
-        "update_time": "2217-06-26 21:20:07"
+        "update_time": "2217-06-26T21:20:07Z"
       },
       "stream_descriptor": {
         "name": "organizations"

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

@@ -144,10 +144,13 @@
       "stream": {
         "name": "organizations",
         "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["update_time"]
       },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite"
+      "sync_mode": "incremental",
+      "cursor_field": ["update_time"],
+      "destination_sync_mode": "append"
     },
     {
       "stream": {