Source Twilio: implement slicing (airbytehq#18423)

* airbytehq#804 source twilio: implement slicing

* airbytehq#804 source twilio - format code

* airbytehq#804 source twilio - implement slicing

* airbytehq#804 source twilio: configure incremental streams

* airbytehq#804 fix expected records

* airbytehq#804 extend Calls stream schema

* airbytehq#804 source twilio: allow configurable slice for SATs

* auto-bump connector version

Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -1111,7 +1111,7 @@
 - name: Twilio
   sourceDefinitionId: b9dc6155-672e-42ea-b10d-9f1f1fb95ab1
   dockerRepository: airbyte/source-twilio
-  dockerImageTag: 0.1.12
+  dockerImageTag: 0.1.13
   documentationUrl: https://docs.airbyte.com/integrations/sources/twilio
   icon: twilio.svg
   sourceType: api

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -11610,7 +11610,7 @@
         oauthFlowOutputParameters:
         - - "token"
         - - "key"
-- dockerImage: "airbyte/source-twilio:0.1.12"
+- dockerImage: "airbyte/source-twilio:0.1.13"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/twilio"
     connectionSpecification:

airbyte-integrations/connectors/source-twilio/Dockerfile

@@ -12,5 +12,5 @@ COPY main.py ./
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.12
+LABEL io.airbyte.version=0.1.13
 LABEL io.airbyte.name=airbyte/source-twilio

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

@@ -16,6 +16,8 @@ tests:
       configured_catalog_path: "integration_tests/no_empty_streams_catalog.json"
       expect_records:
         path: "integration_tests/expected_records.txt"
+      empty_streams: ["alerts"]
+      timeout_seconds: 600
   incremental:
     - config_path: "secrets/config.json"
       # usage records stream produces and error if cursor date gte than current date

airbyte-integrations/connectors/source-twilio/source_twilio/schemas/calls.json

@@ -109,6 +109,12 @@
         },
         "streams": {
           "type": ["null", "string"]
+        },
+        "user_defined_message_subscriptions": {
+          "type": ["null", "string"]
+        },
+        "user_defined_messages": {
+          "type": ["null", "string"]
         }
       }
     }

airbyte-integrations/connectors/source-twilio/source_twilio/source.py

@@ -69,6 +69,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             "authenticator": auth,
             "start_date": config["start_date"],
             "lookback_window": config.get("lookback_window", 0),
+            "slice_step_map": config.get("slice_step_map", {}),
         }
 
         # Fix for `Date range specified in query is partially or entirely outside of retention window of 400 days`

airbyte-integrations/connectors/source-twilio/unit_tests/test_streams.py

@@ -2,14 +2,16 @@
 # Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 #
 
+from contextlib import nullcontext
 from unittest.mock import patch
 
+import pendulum
 import pytest
 import requests
 from airbyte_cdk.sources.streams.http import HttpStream
 from source_twilio.auth import HttpBasicAuthenticator
 from source_twilio.source import SourceTwilio
-from source_twilio.streams import Accounts, Addresses, Calls, DependentPhoneNumbers, MessageMedia, Messages, UsageTriggers
+from source_twilio.streams import Accounts, Addresses, Alerts, Calls, DependentPhoneNumbers, MessageMedia, TwilioNestedStream, UsageTriggers
 
 TEST_CONFIG = {
     "account_sid": "airbyte.io",
@@ -135,29 +137,25 @@ class TestIncrementalTwilioStream:
     CONFIG.pop("auth_token")
 
     @pytest.mark.parametrize(
-        "stream_cls, expected",
-        [
-            (Calls, "EndTime>"),
-        ],
-    )
-    def test_incremental_filter_field(self, stream_cls, expected):
-        stream = stream_cls(**self.CONFIG)
-        result = stream.incremental_filter_field
-        assert result == expected
-
-    @pytest.mark.parametrize(
-        "stream_cls, next_page_token, expected",
+        "stream_cls, stream_slice, next_page_token, expected",
         [
             (
                 Calls,
+                {"EndTime>": "2022-01-01", "EndTime<": "2022-01-02"},
                 {"Page": "2", "PageSize": "1000", "PageToken": "PAAD42931b949c0dedce94b2f93847fdcf95"},
-                {"EndTime>": "2022-01-01", "Page": "2", "PageSize": "1000", "PageToken": "PAAD42931b949c0dedce94b2f93847fdcf95"},
+                {
+                    "EndTime>": "2022-01-01",
+                    "EndTime<": "2022-01-02",
+                    "Page": "2",
+                    "PageSize": "1000",
+                    "PageToken": "PAAD42931b949c0dedce94b2f93847fdcf95",
+                },
             ),
         ],
     )
-    def test_request_params(self, stream_cls, next_page_token, expected):
+    def test_request_params(self, stream_cls, stream_slice, next_page_token, expected):
         stream = stream_cls(**self.CONFIG)
-        result = stream.request_params(stream_state=None, next_page_token=next_page_token)
+        result = stream.request_params(stream_state=None, stream_slice=stream_slice, next_page_token=next_page_token)
         assert result == expected
 
     @pytest.mark.parametrize(
@@ -172,6 +170,33 @@ def test_read_records(self, stream_cls, record, expected):
             result = stream.read_records(sync_mode=None)
             assert list(result) == expected
 
+    @pytest.mark.parametrize(
+        "stream_cls, parent_cls_records, extra_slice_keywords",
+        [
+            (Calls, [{"subresource_uris": {"calls": "123"}}, {"subresource_uris": {"calls": "124"}}], ["subresource_uri"]),
+            (Alerts, [{}], []),
+        ],
+    )
+    def test_stream_slices(self, mocker, stream_cls, parent_cls_records, extra_slice_keywords):
+        stream = stream_cls(
+            authenticator=TEST_CONFIG.get("authenticator"), start_date=pendulum.now().subtract(months=13).to_iso8601_string()
+        )
+        expected_slices = 2 * len(parent_cls_records)  # 2 per year slices per each parent slice
+        if isinstance(stream, TwilioNestedStream):
+            slices_mock_context = mocker.patch.object(stream.parent_stream_instance, "stream_slices", return_value=[{}])
+            records_mock_context = mocker.patch.object(stream.parent_stream_instance, "read_records", return_value=parent_cls_records)
+        else:
+            slices_mock_context, records_mock_context = nullcontext(), nullcontext()
+        with slices_mock_context:
+            with records_mock_context:
+                slices = list(stream.stream_slices(sync_mode="incremental"))
+        assert len(slices) == expected_slices
+        for slice_ in slices:
+            if isinstance(stream, TwilioNestedStream):
+                for kw in extra_slice_keywords:
+                    assert kw in slice_
+            assert slice_[stream.lower_boundary_filter_field] <= slice_[stream.upper_boundary_filter_field]
+
 
 class TestTwilioNestedStream:
 
@@ -205,20 +230,14 @@ def test_media_exist_validation(self, stream_cls, expected):
                 [{"subresource_uris": {"addresses": "123"}, "sid": "123", "account_sid": "456"}],
                 [{"sid": "123", "account_sid": "456"}],
             ),
-            (
-                MessageMedia,
-                Messages,
-                [{"subresource_uris": {"media": "1234"}, "num_media": "1", "sid": "123", "account_sid": "456"}],
-                [{"subresource_uri": "1234"}],
-            ),
         ],
     )
     def test_stream_slices(self, stream_cls, parent_stream, record, expected):
         stream = stream_cls(**self.CONFIG)
         with patch.object(Accounts, "read_records", return_value=record):
             with patch.object(parent_stream, "stream_slices", return_value=record):
                 with patch.object(parent_stream, "read_records", return_value=record):
-                    result = stream.stream_slices()
+                    result = stream.stream_slices(sync_mode="full_refresh")
                     assert list(result) == expected
 
 

docs/integrations/sources/twilio.md

@@ -80,6 +80,7 @@ For more information, see [the Twilio docs for rate limitations](https://support
 
 | Version | Date       | Pull Request                                             | Subject                                                                                                 |
 |:--------|:-----------|:---------------------------------------------------------|:--------------------------------------------------------------------------------------------------------|
+| 0.1.13  | 2022-10-25 | [18423](https://github.com/airbytehq/airbyte/pull/18423) | Implement datetime slicing for streams supporting incremental syncs                                     |
 | 0.1.11  | 2022-09-30 | [17478](https://github.com/airbytehq/airbyte/pull/17478) | Add lookback_window parameters                                                                          |
 | 0.1.10  | 2022-09-29 | [17410](https://github.com/airbytehq/airbyte/pull/17410) | Migrate to per-stream states                                                                            |
 | 0.1.9   | 2022-09-26 | [17134](https://github.com/airbytehq/airbyte/pull/17134) | Add test data for Message Media and Conferences                                                         |