fix(source-intercom): conversation parts stream moved to per-partiiti…

…on state (#46658)
airbytehq · Oct 22, 2024 · f883bda · f883bda
1 parent fb8063b
commit f883bda
Show file tree

Hide file tree

Showing 5 changed files with 88 additions and 77 deletions.
diff --git a/airbyte-integrations/connectors/source-intercom/metadata.yaml b/airbyte-integrations/connectors/source-intercom/metadata.yaml
@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: d8313939-3782-41b0-be29-b3ca20d8dd3a
-  dockerImageTag: 0.7.5
+  dockerImageTag: 0.8.0
   dockerRepository: airbyte/source-intercom
   documentationUrl: https://docs.airbyte.com/integrations/sources/intercom
   githubIssueLabel: source-intercom

diff --git a/airbyte-integrations/connectors/source-intercom/pyproject.toml b/airbyte-integrations/connectors/source-intercom/pyproject.toml
@@ -3,7 +3,7 @@ requires = [ "poetry-core>=1.0.0",]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "0.7.5"
+version = "0.8.0"
 name = "source-intercom"
 description = "Source implementation for Intercom Yaml."
 authors = [ "Airbyte <[email protected]>",]

diff --git a/airbyte-integrations/connectors/source-intercom/source_intercom/manifest.yaml b/airbyte-integrations/connectors/source-intercom/source_intercom/manifest.yaml
@@ -467,8 +467,8 @@ definitions:
         record_filter:
           condition:
             "{{ record['updated_at'] >= ( stream_state.get('prior_state',
-            {}).get('updated_at', 0) if stream_state else stream_slice.get('prior_state',
-            {}).get('updated_at', 0) ) }}"
+            {}).get('updated_at', 0) - (config.get('lookback_window', 0) * 86400) if stream_state else stream_slice.get('prior_state',
+            {}).get('updated_at', 0) ) - (config.get('lookback_window', 0) * 86400) }}"
   segments:
     description: "https://developers.intercom.com/intercom-api-reference/reference#list-segments"
     $ref: "#/definitions/stream_semi_incremental"
@@ -740,7 +740,7 @@ definitions:
         record_filter:
           condition:
             "{{ record['updated_at'] >= stream_state.get('prior_state', {}).get('updated_at',
-            0) }}"
+            0) - (config.get('lookback_window', 0) * 86400)}}"
   conversation_parts:
     $ref: "#/definitions/substream_semi_incremental"
     incremental_sync:
@@ -1008,8 +1008,8 @@ definitions:
           description: "https://developers.intercom.com/intercom-api-reference/reference/pagination-sorting-search#pagination"
           condition:
             "{{ record['updated_at'] >= ( stream_state.get('prior_state',
-            {}).get('updated_at', 0) if stream_state else stream_slice.get('prior_state',
-            {}).get('updated_at', 0) ) }}"
+            {}).get('updated_at', 0) - (config.get('lookback_window', 0) * 86400) if stream_state else stream_slice.get('prior_state',
+            {}).get('updated_at', 0) ) - (config.get('lookback_window', 0) * 86400)}}"
       paginator:
         type: "DefaultPaginator"
         url_base: "#/definitions/requester/url_base"
@@ -2318,6 +2318,7 @@ definitions:
       cursor_datetime_formats:
         - "%s"
       datetime_format: "%s"
+      lookback_window: "P{{ config.get('lookback_window', 0) }}D"
       cursor_granularity: "PT1S"
       step: "P{{ config.get('activity_logs_time_step', 30) }}D"
       start_datetime:

diff --git a/airbyte-integrations/connectors/source-intercom/source_intercom/spec.json b/airbyte-integrations/connectors/source-intercom/source_intercom/spec.json
@@ -45,6 +45,15 @@
         "description": "Set lower value in case of failing long running sync of Activity Logs stream.",
         "examples": [30, 10, 5],
         "order": 3
+      },
+      "lookback_window": {
+        "title": "Lookback window",
+        "description": "The number of days to shift the state value backward for record sync",
+        "examples": [60],
+        "default": 0,
+        "minimum": 0,
+        "type": "integer",
+        "order": 4
       }
     }
   },