Source twilio: implement rolling windows

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

@@ -60,8 +60,8 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
                 config["auth_token"],
             ),
         )
-        full_refresh_stream_kwargs = {"authenticator": auth}
-        incremental_stream_kwargs = {"authenticator": auth, "start_date": config["start_date"]}
+        full_refresh_stream_kwargs = {"authenticator": auth, "pagesize": config["page_size"]}
+        incremental_stream_kwargs = {"authenticator": auth, "start_date": config["start_date"], "lookback": config["lookback"], "pagesize": config["page_size"]}
 
         streams = [
             Accounts(**full_refresh_stream_kwargs),

airbyte-integrations/connectors/source-twilio/source_twilio/spec.json

@@ -4,30 +4,63 @@
     "$schema": "http://json-schema.org/draft-07/schema#",
     "title": "Twilio Spec",
     "type": "object",
-    "required": ["account_sid", "auth_token", "start_date"],
+    "required": [
+      "account_sid",
+      "auth_token",
+      "start_date"
+    ],
     "additionalProperties": false,
     "properties": {
       "account_sid": {
         "title": "Account ID",
         "description": "Twilio account SID",
         "airbyte_secret": true,
-        "type": "string"
+        "type": "string",
+        "order": 1
       },
       "auth_token": {
         "title": "Auth Token",
         "description": "Twilio Auth Token.",
         "airbyte_secret": true,
-        "type": "string"
+        "type": "string",
+        "order": 2
       },
       "start_date": {
         "title": "Replication Start Date",
         "description": "UTC date and time in the format 2020-10-01T00:00:00Z. Any data before this date will not be replicated.",
         "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$",
-        "examples": ["2020-10-01T00:00:00Z"],
-        "type": "string"
+        "examples": [
+          "2020-10-01T00:00:00Z"
+        ],
+        "type": "string",
+        "order": 3
+      },
+      "lookback": {
+        "title": "Lookback",
+        "description": "format in minutes. Any data between sync time and sync time minus lookback setting in minutes will be replicated.",
+        "examples": [
+          "60"
+        ],
+        "default": 0,
+        "type": "string",
+        "order": 4
+      },
+      "page_size": {
+        "title": "Page Size",
+        "description": "How many resources to return in each list page. The default is 50, and the maximum is 1000.",
+        "examples": [
+          "500"
+        ],
+        "minimum": 50,
+        "maximum": 1000,
+        "default": 50,
+        "type": "string",
+        "order": 5
       }
     }
   },
   "supportsIncremental": true,
-  "supported_destination_sync_modes": ["append"]
-}
+  "supported_destination_sync_modes": [
+    "append"
+  ]
+}

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

@@ -2,15 +2,21 @@
 # Copyright (c) 2021 Airbyte, Inc., all rights reserved.
 #
 
+
 from abc import ABC, abstractmethod
-from typing import Any, Iterable, Mapping, MutableMapping, Optional
+from asyncio.log import logger
+from datetime import datetime, timedelta
+from distutils.command.config import config
+from typing import Any, Iterable, List, Mapping, MutableMapping, Optional
 from urllib.parse import parse_qsl, urlparse
 
 import pendulum
 import requests
 from airbyte_cdk.models import SyncMode
+from airbyte_cdk.sources.streams import IncrementalMixin
 from airbyte_cdk.sources.streams.http import HttpStream
 from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer
+from typing_extensions import Self
 
 TWILIO_API_URL_BASE = "https://api.twilio.com"
 TWILIO_API_URL_BASE_VERSIONED = f"{TWILIO_API_URL_BASE}/2010-04-01/"
@@ -20,9 +26,12 @@
 class TwilioStream(HttpStream, ABC):
     url_base = TWILIO_API_URL_BASE
     primary_key = "sid"
-    page_size = 100
     transformer: TypeTransformer = TypeTransformer(TransformConfig.DefaultSchemaNormalization | TransformConfig.CustomSchemaNormalization)
-
+
+    def __init__(self, pagesize: int = 50, **kwargs):
+        super().__init__(**kwargs)
+        self._page_size = pagesize
+
     @property
     def data_field(self):
         return self.name
@@ -41,6 +50,7 @@ def next_page_token(self, response: requests.Response) -> Optional[Mapping[str,
         stream_data = response.json()
         next_page_uri = stream_data.get("next_page_uri")
         if next_page_uri:
+            print (next_page_uri)
             next_url = urlparse(next_page_uri)
             next_page_params = dict(parse_qsl(next_url.query))
             return next_page_params
@@ -67,12 +77,11 @@ def backoff_time(self, response: requests.Response) -> Optional[float]:
         backoff_time = response.headers.get("Retry-After")
         if backoff_time is not None:
           return float(backoff_time)
-
-    def request_params(
-        self, stream_state: Mapping[str, Any], next_page_token: Mapping[str, Any] = None, **kwargs
-    ) -> MutableMapping[str, Any]:
+
+
+    def request_params(self, stream_state: Mapping[str, Any], next_page_token: Mapping[str, Any] = None, **kwargs) -> MutableMapping[str, Any]:
         params = super().request_params(stream_state=stream_state, next_page_token=next_page_token, **kwargs)
-        params["PageSize"] = self.page_size
+        params["PageSize"] = self._page_size  
         if next_page_token:
             params.update(**next_page_token)
         return params
@@ -90,15 +99,18 @@ def custom_transform_function(original_value: Any, field_schema: Mapping[str, An
                 # is no need in transforming anything.
                 pass
         return original_value
+
 
 
-class IncrementalTwilioStream(TwilioStream, ABC):
+class IncrementalTwilioStream(TwilioStream, IncrementalMixin):
     cursor_field = "date_updated"
     time_filter_template = "%Y-%m-%dT%H:%M:%SZ"
-
-    def __init__(self, start_date: str = None, **kwargs):
+    
+    def __init__(self, start_date: str = None, lookback: int = 0, **kwargs):
         super().__init__(**kwargs)
-        self._start_date = start_date
+        self._start_date = start_date  
+        self._lookback = lookback
+        self._cursor_value = None
 
     @property
     @abstractmethod
@@ -107,28 +119,29 @@ def incremental_filter_field(self) -> str:
         return: date filter query parameter name
         """
 
-    def get_updated_state(self, current_stream_state: MutableMapping[str, Any], latest_record: Mapping[str, Any]) -> Mapping[str, Any]:
-        """
-        Return the latest state by comparing the cursor value in the latest record with the stream's most recent state object
-        and returning an updated state object.
-        """
-        latest_benchmark = pendulum.parse(latest_record[self.cursor_field], strict=False).strftime(self.time_filter_template)
-        if current_stream_state.get(self.cursor_field):
-            return {self.cursor_field: max(latest_benchmark, current_stream_state[self.cursor_field])}
-        return {self.cursor_field: latest_benchmark}
+    @property
+    def state(self) -> Mapping[str, Any]:
+        if self._cursor_value:
+            return {self.cursor_field: self._cursor_value.strftime(self.time_filter_template)}
+        else:
+            return {self.cursor_field: self._start_date.strftime(self.time_filter_template)}
+
+    @state.setter
+    def state(self, value: Mapping[str, Any]):
+       self._cursor_value = datetime.strptime(value[self.cursor_field], self.time_filter_template)
 
     def request_params(self, stream_state: Mapping[str, Any], **kwargs) -> MutableMapping[str, Any]:
         params = super().request_params(stream_state=stream_state, **kwargs)
         start_date = stream_state.get(self.cursor_field) or self._start_date
         if start_date:
-            params.update({self.incremental_filter_field: pendulum.parse(start_date, strict=False).strftime(self.time_filter_template)})
+            params.update({self.incremental_filter_field: (pendulum.parse(start_date, strict=False)-timedelta(minutes=int(self._lookback))).strftime(self.time_filter_template)})
         return params
 
-    def read_records(self, stream_state: Mapping[str, Any] = None, **kwargs):
-        stream_state = stream_state or {}
-        records = super().read_records(stream_state=stream_state, **kwargs)
-        for record in records:
-            record[self.cursor_field] = pendulum.parse(record[self.cursor_field], strict=False).strftime(self.time_filter_template)
+    def read_records(self, *args, **kwargs) -> Iterable[Mapping[str, Any]]:
+        for record in super().read_records(*args, **kwargs):
+            if self._cursor_value:
+                latest_record_date = datetime.strptime(record[self.cursor_field], self.time_filter_template)
+                self._cursor_value = max(self._cursor_value, latest_record_date)
             yield record
 
 
@@ -333,6 +346,7 @@ class Messages(TwilioNestedStream, IncrementalTwilioStream):
     parent_stream = Accounts
     incremental_filter_field = "DateSent>"
     cursor_field = "date_sent"
+    page_size = config
 
 
 class MessageMedia(TwilioNestedStream, IncrementalTwilioStream):

docker-compose.yaml

@@ -1,6 +1,7 @@
 version: "3.7"
 #https://github.com/compose-spec/compose-spec/blob/master/spec.md#using-extensions-as-fragments
-x-logging: &default-logging
+x-logging:
+  &default-logging
   options:
     max-size: "100m"
     max-file: "5"
@@ -48,6 +49,8 @@ services:
       - POSTGRES_USER=${DATABASE_USER}
     volumes:
       - db:/var/lib/postgresql/data
+    ports:
+      - 5432:5432
   scheduler:
     image: airbyte/scheduler:${VERSION}
     logging: *default-logging