Source Twilio: Migrate connector from GA (Python) to GA (YAML)

airbyte-cdk/python/airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -379,6 +379,9 @@ definitions:
       datetime_format:
         description: The format of the datetime
         type: string
+      cursor_datetime_format:
+        description: The format string used to transform the datetime value in the API response into the format used by the cursor
+        type: string
       cursor_granularity:
         description: Smallest increment the datetime_format has (ISO 8601 duration) that is used to ensure the start of a slice does not overlap with the end of the previous one
         type: string

airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

@@ -44,6 +44,7 @@ class DatetimeBasedCursor(StreamSlicer):
         partition_field_start (Optional[str]): partition start time field
         partition_field_end (Optional[str]): stream slice end time field
         lookback_window (Optional[InterpolatedString]): how many days before start_datetime to read data for (ISO8601 duration)
+        cursor_datetime_format (str): the format string used to transform the datetime value in the API response into the format used by the cursor
     """
 
     start_datetime: Union[MinMaxDatetime, str]
@@ -61,6 +62,7 @@ class DatetimeBasedCursor(StreamSlicer):
     partition_field_start: Optional[str] = None
     partition_field_end: Optional[str] = None
     lookback_window: Optional[Union[InterpolatedString, str]] = None
+    cursor_datetime_format: str = None
 
     def __post_init__(self, parameters: Mapping[str, Any]):
         if not isinstance(self.start_datetime, MinMaxDatetime):
@@ -100,6 +102,8 @@ def update_cursor(self, stream_slice: StreamSlice, last_record: Optional[Record]
         stream_slice_value = stream_slice.get(self.cursor_field.eval(self.config))
         stream_slice_value_end = stream_slice.get(self.partition_field_end.eval(self.config))
         last_record_value = last_record.get(self.cursor_field.eval(self.config)) if last_record else None
+        if last_record_value and self.cursor_datetime_format:
+            last_record_value = self._format_datetime(self._parser.parse(last_record_value, self.cursor_datetime_format, self._timezone))
         cursor = None
         if stream_slice_value and last_record_value:
             cursor = max(stream_slice_value, last_record_value)

airbyte-cdk/python/airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -401,6 +401,10 @@ class DatetimeBasedCursor(BaseModel):
         description="The location of the value on a record that will be used as a bookmark during sync",
     )
     datetime_format: str = Field(..., description="The format of the datetime")
+    cursor_datetime_format: Optional[str] = Field(
+        None,
+        description="The format string used to transform the datetime value in the API response into the format used by the cursor",
+    )
     cursor_granularity: str = Field(
         ...,
         description="Smallest increment the datetime_format has (ISO 8601 duration) that is used to ensure the start of a slice does not overlap with the end of the previous one",

airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -412,6 +412,7 @@ def create_datetime_based_cursor(self, model: DatetimeBasedCursorModel, config:
             cursor_field=model.cursor_field,
             cursor_granularity=model.cursor_granularity,
             datetime_format=model.datetime_format,
+            cursor_datetime_format=model.cursor_datetime_format,
             end_datetime=end_datetime,
             start_datetime=start_datetime,
             step=model.step,

airbyte-cdk/python/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -2,6 +2,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+import copy
 import json
 import logging
 from dataclasses import InitVar, dataclass, field
@@ -334,7 +335,7 @@ def parse_response(
         records = self.record_selector.select_records(
             response=response, stream_state=self.state, stream_slice=stream_slice, next_page_token=next_page_token
         )
-        self._last_records = records
+        self._last_records = copy.deepcopy(records)
         return records
 
     @property

airbyte-cdk/python/airbyte_cdk/sources/declarative/transformations/add_fields.py

@@ -109,10 +109,11 @@ def transform(
         stream_state: Optional[StreamState] = None,
         stream_slice: Optional[StreamSlice] = None,
     ) -> Record:
-        kwargs = {"record": record, "stream_state": stream_state, "stream_slice": stream_slice}
-        for parsed_field in self._parsed_fields:
-            value = parsed_field.value.eval(config, **kwargs)
-            dpath.util.new(record, parsed_field.path, value)
+        if isinstance(record, Mapping):
+            kwargs = {"record": record, "stream_state": stream_state, "stream_slice": stream_slice}
+            for parsed_field in self._parsed_fields:
+                value = parsed_field.value.eval(config, **kwargs)
+                dpath.util.new(record, parsed_field.path, value)
 
         return record
 

airbyte-cdk/python/unit_tests/sources/declarative/incremental/test_datetime_based_cursor.py

@@ -394,53 +394,67 @@ def test_stream_slices(
 
 
 @pytest.mark.parametrize(
-    "test_name, previous_cursor, stream_slice, last_record, expected_state",
+    "test_name, previous_cursor, stream_slice, last_record, expected_state, cursor_datetime_format",
     [
-        ("test_update_cursor_no_state_no_record", None, {}, None, {}),
+        ("test_update_cursor_no_state_no_record", None, {}, None, {}, None),
         (
             "test_update_cursor_with_state_no_record",
             None,
             {cursor_field: "2021-01-02T00:00:00.000000+0000"},
             None,
             {cursor_field: "2021-01-02T00:00:00.000000+0000"},
+            None
         ),
         (
             "test_update_cursor_with_state_equals_record",
             None,
             {cursor_field: "2021-01-02T00:00:00.000000+0000"},
             {cursor_field: "2021-01-02T00:00:00.000000+0000"},
             {cursor_field: "2021-01-02T00:00:00.000000+0000"},
+            None
         ),
         (
             "test_update_cursor_with_state_greater_than_record",
             None,
             {cursor_field: "2021-01-03T00:00:00.000000+0000"},
             {cursor_field: "2021-01-02T00:00:00.000000+0000"},
             {cursor_field: "2021-01-03T00:00:00.000000+0000"},
+            None
         ),
         (
             "test_update_cursor_with_state_less_than_record",
             None,
             {cursor_field: "2021-01-02T00:00:00.000000+0000"},
             {cursor_field: "2021-01-03T00:00:00.000000+0000"},
             {cursor_field: "2021-01-03T00:00:00.000000+0000"},
+            None
         ),
         (
             "test_update_cursor_with_state_less_than_previous_cursor",
             "2021-01-03T00:00:00.000000+0000",
             {cursor_field: "2021-01-02T00:00:00.000000+0000"},
             {},
             {cursor_field: "2021-01-03T00:00:00.000000+0000"},
+            None
+        ),
+        (
+            "test_update_cursor_last_record_different_format",
+            "2021-01-03T00:00:00.000000+0000",
+            {},
+            {cursor_field: "Wed, 06 Jan 2021 00:00:00 +0000"},
+            {cursor_field: "2021-01-06T00:00:00.000000+0000"},
+            "%a, %d %b %Y %H:%M:%S %z"
         ),
     ],
 )
-def test_update_cursor(test_name, previous_cursor, stream_slice, last_record, expected_state):
+def test_update_cursor(test_name, previous_cursor, stream_slice, last_record, expected_state, cursor_datetime_format):
     slicer = DatetimeBasedCursor(
         start_datetime=MinMaxDatetime(datetime="2021-01-01T00:00:00.000000+0000", parameters={}),
         end_datetime=MinMaxDatetime(datetime="2021-01-10T00:00:00.000000+0000", parameters={}),
         step="P1D",
         cursor_field=InterpolatedString(string=cursor_field, parameters={}),
         datetime_format=datetime_format,
+        cursor_datetime_format=cursor_datetime_format,
         cursor_granularity=cursor_granularity,
         lookback_window=InterpolatedString(string="0d", parameters={}),
         config=config,

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

@@ -2164,6 +2164,7 @@
     hosts:
       - api.twilio.com
       - monitor.twilio.com
+      - studio.twilio.com
 - name: Twilio Taskrouter
   sourceDefinitionId: 2446953b-b794-429b-a9b3-c821ba992a48
   dockerRepository: airbyte/source-twilio-taskrouter

airbyte-integrations/connectors/source-twilio/.dockerignore

@@ -3,4 +3,3 @@
 !main.py
 !source_twilio
 !setup.py
-!secrets

airbyte-integrations/connectors/source-twilio/requirements.txt

@@ -1,2 +1,2 @@
 -e ../../bases/connector-acceptance-test
--e .
+-e .

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

@@ -20,7 +20,7 @@
     author_email="[email protected]",
     packages=find_packages(),
     install_requires=MAIN_REQUIREMENTS,
-    package_data={"": ["*.json", "schemas/*.json", "schemas/shared/*.json"]},
+    package_data={"": ["*.json", "*.yaml", "schemas/*.json", "schemas/shared/*.json"]},
     extras_require={
         "tests": TEST_REQUIREMENTS,
     },

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

@@ -1,26 +1,7 @@
-"""
-MIT License
+#
+# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+#
 
-Copyright (c) 2020 Airbyte
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
-"""
 
 from .source import SourceTwilio
 

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

@@ -0,0 +1,51 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any, Iterable, Mapping, Optional
+
+from airbyte_cdk.models import SyncMode
+from airbyte_cdk.sources.declarative.stream_slicers import CartesianProductStreamSlicer
+from airbyte_cdk.sources.declarative.types import Record, StreamSlice
+
+
+@dataclass
+class NestedStateCartesianProductStreamSlicer(CartesianProductStreamSlicer):
+    """
+    [Low-Code Custom Component] NestedStateCartesianProductStreamSlicer
+    https://github.com/airbytehq/airbyte/issues/22873
+
+    Some streams require support of nested state:
+    {
+      "recordings": {
+        "account_id": {
+          "date_created": "2023-01-19T23:30:52Z"
+        }
+      }
+    }
+    """
+
+    def __post_init__(self, options: Mapping[str, Any]):
+        self._cursor = {}
+        self._options = options
+
+    def get_stream_state(self) -> Mapping[str, Any]:
+        return self._cursor
+
+    def stream_slices(self, sync_mode: SyncMode, stream_state: Mapping[str, Any]) -> Iterable[Mapping[str, Any]]:
+        record_slicer, datetime_slicer = self.stream_slicers
+        for record_slice in record_slicer.stream_slices(sync_mode, stream_state):
+            record_id = record_slice["record"]["sid"]
+            datetime_slicer._cursor = None
+            for datetime_slice in datetime_slicer.stream_slices(sync_mode, stream_state.get(record_id)):
+                yield record_slice | datetime_slice
+
+    def update_cursor(self, stream_slice: StreamSlice, last_record: Optional[Record] = None):
+        datetime_slicer = self.stream_slicers[1]
+        datetime_slicer.update_cursor(stream_slice, last_record)
+        if last_record:
+            record_id = stream_slice["record"]["sid"]
+            self._cursor.setdefault(record_id, {}).update(datetime_slicer.get_stream_state())
+        else:
+            self._cursor = stream_slice