Source Sentry: add incremental sync

airbyte-integrations/connectors/source-sentry/Dockerfile

@@ -34,5 +34,5 @@ COPY source_sentry ./source_sentry
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.7
+LABEL io.airbyte.version=0.1.8
 LABEL io.airbyte.name=airbyte/source-sentry

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

@@ -1,18 +1,33 @@
 connector_image: airbyte/source-sentry:dev
-tests:
+acceptance_tests:
   spec:
-    - spec_path: "source_sentry/spec.json"
+    tests:
+      - spec_path: "source_sentry/spec.json"
   connection:
-    - config_path: "secrets/config.json"
-      status: "succeed"
-    - config_path: "integration_tests/invalid_config.json"
-      status: "failed"
+    tests:
+      - config_path: "secrets/config.json"
+        status: "succeed"
+      - config_path: "integration_tests/invalid_config.json"
+        status: "failed"
   discovery:
-    - config_path: "secrets/config.json"
+    tests:
+      - config_path: "secrets/config.json"
   basic_read:
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog.json"
-      empty_streams: ["issues", "events"]
+    tests:
+      - config_path: "secrets/config.json"
+        configured_catalog_path: "integration_tests/configured_catalog.json"
+        empty_streams:
+          - name: issues
+            bypass_reason: "unable to populate"
+          - name: events
+            bypass_reason: "unable to populate"
+  incremental:
+    tests:
+      - config_path: "secrets/config.json"
+        configured_catalog_path: "integration_tests/configured_catalog.json"
+        future_state:
+          future_state_path: "integration_tests/abnormal_state.json"
   full_refresh:
-    - config_path: "secrets/config.json"
-      configured_catalog_path: "integration_tests/configured_catalog.json"
+    tests:
+      - config_path: "secrets/config.json"
+        configured_catalog_path: "integration_tests/configured_catalog.json"

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

@@ -0,0 +1,13 @@
+[
+  {
+    "type": "STREAM",
+    "stream": {
+      "stream_state": {
+        "dateCreated": "2100-01-01T00:00:00.0Z"
+      },
+      "stream_descriptor": {
+        "name": "projects"
+      }
+    }
+  }
+]

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

@@ -28,12 +28,12 @@
       }
     },
     {
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite",
+      "sync_mode": "incremental",
+      "destination_sync_mode": "append",
       "stream": {
         "name": "projects",
         "json_schema": {},
-        "supported_sync_modes": ["full_refresh"]
+        "supported_sync_modes": ["full_refresh", "incremental"]
       }
     }
   ]

airbyte-integrations/connectors/source-sentry/source_sentry/schemas/events.json

@@ -20,7 +20,8 @@
       }
     },
     "dateCreated": {
-      "type": "string"
+      "type": "string",
+      "format": "date-time"
     },
     "user": {
       "type": ["null", "object"],

airbyte-integrations/connectors/source-sentry/source_sentry/streams.py

@@ -6,7 +6,9 @@
 from abc import ABC
 from typing import Any, Iterable, Mapping, MutableMapping, Optional
 
+import pendulum
 import requests
+from airbyte_cdk.sources.streams import IncrementalMixin
 from airbyte_cdk.sources.streams.http import HttpStream
 
 
@@ -62,11 +64,40 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
         yield from response.json()
 
 
-class Events(SentryStreamPagination):
+class SentryIncremental(SentryStreamPagination, IncrementalMixin):
+    def filter_by_state(self, stream_state: Mapping[str, Any] = None, record: Mapping[str, Any] = None) -> Iterable:
+        """
+        Endpoint does not provide query filtering params, but they provide us
+        cursor field in most cases, so we used that as incremental filtering
+        during the parsing.
+        """
+        start_date = "1900-01-01T00:00:00.0Z"
+        if pendulum.parse(record[self.cursor_field]) >= pendulum.parse((stream_state or {}).get(self.cursor_field, start_date)):
+            yield record
+
+    def parse_response(self, response: requests.Response, stream_state: Mapping[str, Any], **kwargs) -> Iterable[MutableMapping]:
+        json_response = response.json() or []
+
+        for record in json_response:
+            yield from self.filter_by_state(stream_state=stream_state, record=record)
+
+    @property
+    def state(self) -> Mapping[str, Any]:
+        return {self.cursor_field: str(self._cursor_value)}
+
+    @state.setter
+    def state(self, value: Mapping[str, Any]):
+        self._cursor_value = value[self.cursor_field]
+
+
+class Events(SentryIncremental):
     """
     Docs: https://docs.sentry.io/api/events/list-a-projects-events/
     """
 
+    primary_key = "id"
+    cursor_field = "dateCreated"
+
     def __init__(self, organization: str, project: str, **kwargs):
         super().__init__(**kwargs)
         self._organization = organization
@@ -92,11 +123,14 @@ def request_params(
         return params
 
 
-class Issues(SentryStreamPagination):
+class Issues(SentryIncremental):
     """
     Docs: https://docs.sentry.io/api/events/list-a-projects-issues/
     """
 
+    primary_key = "id"
+    cursor_field = "lastSeen"
+
     def __init__(self, organization: str, project: str, **kwargs):
         super().__init__(**kwargs)
         self._organization = organization
@@ -122,11 +156,14 @@ def request_params(
         return params
 
 
-class Projects(SentryStreamPagination):
+class Projects(SentryIncremental):
     """
     Docs: https://docs.sentry.io/api/projects/list-your-projects/
     """
 
+    primary_key = "id"
+    cursor_field = "dateCreated"
+
     def path(
         self,
         stream_state: Optional[Mapping[str, Any]] = None,

docs/integrations/sources/sentry.md

@@ -45,7 +45,8 @@ The Sentry source connector supports the following [sync modes](https://docs.air
 
 | Version | Date       | Pull Request                                             | Subject                                           |
 |:--------|:-----------|:---------------------------------------------------------|:--------------------------------------------------|
-| 0.1.7   | 2022-09-30 | [17466](https://github.com/airbytehq/airbyte/pull/17466) | Migrate to per-stream states                     |
+| 0.1.8   | 2022-12-20 | [20709](https://github.com/airbytehq/airbyte/pull/20709) | Add incremental sync                              |
+| 0.1.7   | 2022-09-30 | [17466](https://github.com/airbytehq/airbyte/pull/17466) | Migrate to per-stream states                      |
 | 0.1.6   | 2022-08-29 | [16112](https://github.com/airbytehq/airbyte/pull/16112) | Revert back to the Python CDK                     |
 | 0.1.5   | 2022-08-24 | [15911](https://github.com/airbytehq/airbyte/pull/15911) | Bugfix to allowing reading schemas at runtime     |
 | 0.1.4   | 2022-08-19 | [15800](https://github.com/airbytehq/airbyte/pull/15800) | Bugfix to allow reading sentry.yaml at runtime    |