Source Amplitude: Add empty series validation (#17854)

* Add seriaes validation * Updated PR number * auto-bump connector version [ci skip] Co-authored-by: Octavia Squidington III <[email protected]>
airbytehq · Oct 12, 2022 · f21f345 · f21f345
1 parent 5df66cd
commit f21f345
Show file tree

Hide file tree

Showing 6 changed files with 18 additions and 6 deletions.
diff --git a/airbyte-config/init/src/main/resources/seed/source_definitions.yaml b/airbyte-config/init/src/main/resources/seed/source_definitions.yaml
@@ -56,7 +56,7 @@
 - name: Amplitude
   sourceDefinitionId: fa9f58c6-2d03-4237-aaa4-07d75e0c1396
   dockerRepository: airbyte/source-amplitude
-  dockerImageTag: 0.1.15
+  dockerImageTag: 0.1.16
   documentationUrl: https://docs.airbyte.com/integrations/sources/amplitude
   icon: amplitude.svg
   sourceType: api

diff --git a/airbyte-config/init/src/main/resources/seed/source_specs.yaml b/airbyte-config/init/src/main/resources/seed/source_specs.yaml
@@ -1105,7 +1105,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-amplitude:0.1.15"
+- dockerImage: "airbyte/source-amplitude:0.1.16"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/amplitude"
     connectionSpecification:

diff --git a/airbyte-integrations/connectors/source-amplitude/Dockerfile b/airbyte-integrations/connectors/source-amplitude/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.15
+LABEL io.airbyte.version=0.1.16
 LABEL io.airbyte.name=airbyte/source-amplitude
diff --git a/airbyte-integrations/connectors/source-amplitude/source_amplitude/api.py b/airbyte-integrations/connectors/source-amplitude/source_amplitude/api.py
@@ -101,7 +101,7 @@ def get_updated_state(self, current_stream_state: MutableMapping[str, Any], late
         if self.compare_date_template:
             latest_state = pendulum.parse(latest_record[self.cursor_field]).strftime(self.compare_date_template)
         else:
-            latest_state = latest_record[self.cursor_field]
+            latest_state = latest_record.get(self.cursor_field, "")
         return {self.cursor_field: max(latest_state, current_stream_state.get(self.cursor_field, ""))}
 
     def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
@@ -234,7 +234,7 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
         if response_data:
             series = list(map(list, zip(*response_data["series"])))
             for i, date in enumerate(response_data["xValues"]):
-                yield {"date": date, "statistics": dict(zip(response_data["seriesLabels"], series[i]))}
+                yield from [{"date": date, "statistics": dict(zip(response_data["seriesLabels"], series[i]))}] if series else []
 
     def path(self, **kwargs) -> str:
         return f"{self.api_version}/users"

diff --git a/airbyte-integrations/connectors/source-amplitude/unit_tests/test_api.py b/airbyte-integrations/connectors/source-amplitude/unit_tests/test_api.py
@@ -74,6 +74,17 @@ class TestIncrementalStreams:
                 },
                 [{"date": "2021-01-01", "statistics": {0: 1}}, {"date": "2021-01-02", "statistics": {0: 5}}],
             ),
+            (
+                ActiveUsers,
+                {
+                    "xValues": ["2021-01-01", "2021-01-02"],
+                    "series": [],
+                    "seriesCollapsed": [[0]],
+                    "seriesLabels": [0],
+                    "seriesMeta": [{"segmentIndex": 0}],
+                },
+                [],
+            ),
             (
                 AverageSessionLength,
                 {
@@ -86,7 +97,7 @@ class TestIncrementalStreams:
                 [{"date": "2019-05-23", "length": 2}, {"date": "2019-05-24", "length": 6}],
             ),
         ],
-        ids=["ActiveUsers", "AverageSessionLength"],
+        ids=["ActiveUsers", "EmptyActiveUsers", "AverageSessionLength"],
     )
     def test_parse_response(self, requests_mock, stream_cls, data, expected):
         stream = stream_cls("2021-01-01T00:00:00Z")

diff --git a/docs/integrations/sources/amplitude.md b/docs/integrations/sources/amplitude.md
@@ -43,6 +43,7 @@ The Amplitude connector ideally should gracefully handle Amplitude API limitatio
 
 | Version | Date       | Pull Request                                             | Subject                                                                                         |
 |:--------|:-----------|:---------------------------------------------------------|:------------------------------------------------------------------------------------------------|
+| 0.1.16  | 2022-10-11 | [17854](https://github.com/airbytehq/airbyte/pull/17854) | Add empty `series` validation                                                                   |
 | 0.1.15  | 2022-10-03 | [17320](https://github.com/airbytehq/airbyte/pull/17320) | Add validation `start_date` filed if it's in the future                                         |
 | 0.1.14  | 2022-09-28 | [17326](https://github.com/airbytehq/airbyte/pull/17326) | Migrate to per-stream states.                                                                   |
 | 0.1.13  | 2022-08-31 | [16185](https://github.com/airbytehq/airbyte/pull/16185) | Re-release on new `airbyte_cdk==0.1.81`                                                         |