PR feedback and a few small tweaks

Author: brianjlai

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py

@@ -154,7 +154,10 @@ def get_request_body_json(
         return self._get_request_options(RequestOptionType.body_json)
 
     def reset(self, reset_value: Optional[Any] = None) -> None:
-        self.pagination_strategy.reset(reset_value=reset_value)
+        if reset_value:
+            self.pagination_strategy.reset(reset_value=reset_value)
+        else:
+            self.pagination_strategy.reset()
         self._token = self.pagination_strategy.initial_token
 
     def _get_request_options(self, option_type: RequestOptionType) -> MutableMapping[str, Any]:

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/paginators/strategies/offset_increment.py

@@ -66,10 +66,8 @@ def next_page_token(self, response: requests.Response, last_page_size: int, last
             self._offset += last_page_size
             return self._offset
 
-    def reset(self, reset_value: Optional[Any] = None) -> None:
-        if reset_value is None:
-            self._offset = 0
-        elif not isinstance(reset_value, int):
+    def reset(self, reset_value: Optional[Any] = 0) -> None:
+        if not isinstance(reset_value, int):
             raise ValueError(f"Reset value {reset_value} for OffsetIncrement pagination strategy was not an integer")
         else:
             self._offset = reset_value

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

@@ -71,6 +71,9 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._last_record: Optional[Record] = None
         self._parameters = parameters
         self._name = InterpolatedString(self._name, parameters=parameters) if isinstance(self._name, str) else self._name
+
+        # This mapping is used during a resumable full refresh syncs to indicate whether a partition has started syncing
+        # records. Partitions serve as the key and map to True if they already began processing records
         self._synced_partitions: MutableMapping[Any, bool] = dict()
 
     @property  # type: ignore
@@ -357,8 +360,8 @@ def read_records(
                 return
             cursor_value = stream_state.get("next_page_token")
 
-            # The first attempt to read a page for the current partition should have the paginator reset to the current
-            # cursor state which is initially set to the incoming state from the platform
+            # The first attempt to read a page for the current partition should reset the paginator to the current
+            # cursor state which is initially assigned to the incoming state from the platform
             partition_key = self._to_partition_key(_slice.partition)
             if partition_key not in self._synced_partitions:
                 self._synced_partitions[partition_key] = True

airbyte-cdk/python/airbyte_cdk/sources/streams/checkpoint/checkpoint_reader.py

@@ -79,26 +79,43 @@ class CursorBasedCheckpointReader(CheckpointReader):
     """
     CursorBasedCheckpointReader is used by streams that implement a Cursor in order to manage state. This allows the checkpoint
     reader to delegate the complexity of fetching state to the cursor and focus on the iteration over a stream's partitions.
-    Right now only low-code connectors provide cursor implementations, but the logic is extensible to any stream that adheres
-    to the Cursor interface.
+
+    This reader supports the Cursor interface used by Python and low-code sources. Not to be confused with Cursor interface
+    that belongs to the Concurrent CDK.
     """
 
     def __init__(self, cursor: Cursor, stream_slices: Iterable[Optional[Mapping[str, Any]]], read_state_from_cursor: bool = False):
-        # The first attempt of an RFR stream has an empty {} incoming state, but should still make a first attempt to read records
-        # from the first page in next().
         self._cursor = cursor
         self._stream_slices = iter(stream_slices)
+        # read_state_from_cursor is used to delineate that partitions should determine when to stop syncing dynamically according
+        # to the value of the state at runtime. This currently only applies to streams that use resumable full refresh.
         self._read_state_from_cursor = read_state_from_cursor
         self._current_slice: Optional[StreamSlice] = None
         self._finished_sync = False
 
     def next(self) -> Optional[Mapping[str, Any]]:
+        """
+        The next() method returns the next slice of data should be synced for the current stream according to its cursor.
+        This function support iterating over a stream's slices across two dimensions. The first dimension is the stream's
+        partitions like parent records for a substream. The inner dimension is iterating over the cursor value like a
+        date range for incremental streams or a pagination checkpoint for resumable full refresh.
+
+        basic algorithm for iterating through a stream's slices is:
+        1. The first time next() is invoked we get the first partition and return it
+        2. For streams whose cursor value is determined dynamically using stream state
+            1. Get the current state for the current partition
+            2. If the current partition's state is complete, get the next partition
+            3. If the current partition's state is still in progress, emit the next cursor value
+        3. If a stream has processed all partitions, the iterator will raise a StopIteration exception signaling there are no more
+           slices left for extracting more records.
+        """
+
         try:
             if self._current_slice is None:
                 self._current_slice = self._get_next_slice()
                 return self._current_slice
             if self._read_state_from_cursor:
-                state_for_slice = self._cursor.select_state(self._current_slice.get("partition"))
+                state_for_slice = self._cursor.select_state(self._current_slice)
                 if state_for_slice == {"__ab_full_refresh_sync_complete": True}:
                     self._current_slice = self._get_next_slice()
                 else:

airbyte-cdk/python/airbyte_cdk/sources/streams/core.py

@@ -412,6 +412,8 @@ def _get_checkpoint_reader(
                 stream_slices=slices, cursor=cursor, read_state_from_cursor=checkpoint_mode == CheckpointMode.RESUMABLE_FULL_REFRESH
             )
         if checkpoint_mode == CheckpointMode.RESUMABLE_FULL_REFRESH:
+            # Resumable full refresh readers rely on the stream state dynamically being updated during pagination and does
+            # not iterate over a static set of slices.
             return ResumableFullRefreshCheckpointReader(stream_state=stream_state)
         else:
             slices = self.stream_slices(

airbyte-cdk/python/unit_tests/sources/declarative/requesters/paginators/test_offset_increment.py

@@ -73,5 +73,8 @@ def test_offset_increment_reset(reset_value, expected_initial_token, expected_er
         with pytest.raises(expected_error):
             paginator_strategy.reset(reset_value=reset_value)
     else:
-        paginator_strategy.reset(reset_value=reset_value)
+        if reset_value is None:
+            paginator_strategy.reset()
+        else:
+            paginator_strategy.reset(reset_value=reset_value)
         assert paginator_strategy.initial_token == expected_initial_token