Remove most_recent_record arg from Cursor.close_slice (airbytehq#36216)

Author: ambirdsall

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

@@ -3,7 +3,6 @@
 #
 
 from abc import ABC, abstractmethod
-from typing import Optional
 
 from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
 from airbyte_cdk.sources.declarative.types import Record, StreamSlice, StreamState
@@ -35,17 +34,13 @@ def observe(self, stream_slice: StreamSlice, record: Record) -> None:
         pass
 
     @abstractmethod
-    def close_slice(self, stream_slice: StreamSlice, most_recent_record: Optional[Record]) -> None:
+    def close_slice(self, stream_slice: StreamSlice) -> None:
         """
-        Update state based on the stream slice and the latest record. Note that `stream_slice.cursor_slice` and
-        `most_recent_record.associated_slice` are expected to be the same but we make it explicit here that `stream_slice` should be leveraged to
-        update the state.
+        Update state based on the stream slice. Note that `stream_slice.cursor_slice` and `most_recent_record.associated_slice` are expected
+        to be the same but we make it explicit here that `stream_slice` should be leveraged to update the state. We do not pass in the
+        latest record, since cursor instances should maintain the relevant internal state on their own.
 
         :param stream_slice: slice to close
-        :param most_recent_record: the latest record we have received for the slice. This is important to consider because even if the
-          cursor emits a slice, some APIs are not able to enforce the upper boundary. The outcome is that the last_record might have a
-          higher cursor value than the slice upper boundary and if we want to reduce the duplication as much as possible, we need to
-          consider the highest value between the internal cursor, the stream slice upper boundary and the record cursor value.
         """
 
     @abstractmethod

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

@@ -138,7 +138,7 @@ def observe(self, stream_slice: StreamSlice, record: Record) -> None:
         ):
             self._highest_observed_cursor_field_value = record_cursor_value
 
-    def close_slice(self, stream_slice: StreamSlice, _most_recent_record: Optional[Record]) -> None:
+    def close_slice(self, stream_slice: StreamSlice) -> None:
         if stream_slice.partition:
             raise ValueError(f"Stream slice {stream_slice} should not have a partition. Got {stream_slice.partition}.")
         cursor_value_str_by_cursor_value_datetime = dict(

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

@@ -91,20 +91,15 @@ def observe(self, stream_slice: StreamSlice, record: Record) -> None:
             StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice), record
         )
 
-    def close_slice(self, stream_slice: StreamSlice, most_recent_record: Optional[Record]) -> None:
+    def close_slice(self, stream_slice: StreamSlice) -> None:
         try:
-            cursor_most_recent_record = (
-                Record(most_recent_record.data, StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice))
-                if most_recent_record
-                else most_recent_record
-            )
             self._cursor_per_partition[self._to_partition_key(stream_slice.partition)].close_slice(
-                StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice), cursor_most_recent_record
+                StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice)
             )
         except KeyError as exception:
             raise ValueError(
                 f"Partition {str(exception)} could not be found in current state based on the record. This is unexpected because "
-                f"we should only update state for partition that where emitted during `stream_slices`"
+                f"we should only update state for partitions that were emitted during `stream_slices`"
             )
 
     def get_stream_state(self) -> StreamState:

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

@@ -314,7 +314,6 @@ def read_records(
         # Fixing paginator types has a long tail of dependencies
         self._paginator.reset()
 
-        most_recent_record_from_slice = None
         record_generator = partial(
             self._parse_records,
             stream_state=self.state or {},
@@ -326,13 +325,10 @@ def read_records(
             if self.cursor and current_record:
                 self.cursor.observe(_slice, current_record)
 
-            # TODO this is just the most recent record *read*, not necessarily the most recent record *within slice boundaries*; once all
-            # cursors implement a meaningful `observe` method, it can be removed, both from here and the `Cursor.close_slice` method args
-            most_recent_record_from_slice = self._get_most_recent_record(most_recent_record_from_slice, current_record, _slice)
             yield stream_data
 
         if self.cursor:
-            self.cursor.close_slice(_slice, most_recent_record_from_slice)
+            self.cursor.close_slice(_slice)
         return
 
     def _get_most_recent_record(

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

@@ -426,8 +426,7 @@ def test_close_slice(test_name, previous_cursor, stream_slice, observed_records,
     for record_data in observed_records:
         record = Record(record_data, stream_slice)
         cursor.observe(stream_slice, record)
-    last_record = observed_records[-1] if observed_records else None
-    cursor.close_slice(stream_slice, Record(last_record, stream_slice) if last_record else None)
+    cursor.close_slice(stream_slice)
     updated_state = cursor.get_stream_state()
     assert updated_state == expected_state
 
@@ -442,7 +441,7 @@ def test_close_slice_fails_if_slice_has_a_partition():
     )
     stream_slice = StreamSlice(partition={"key": "value"}, cursor_slice={"end_time": "2022-01-01"})
     with pytest.raises(ValueError):
-        cursor.close_slice(stream_slice, Record({"id": 1}, stream_slice))
+        cursor.close_slice(stream_slice)
 
 
 def test_compares_cursor_values_by_chronological_order():
@@ -459,7 +458,7 @@ def test_compares_cursor_values_by_chronological_order():
     cursor.observe(_slice, first_record)
     second_record = Record({cursor_field: "01-03-2023"}, _slice)
     cursor.observe(_slice, second_record)
-    cursor.close_slice(_slice, second_record)
+    cursor.close_slice(_slice)
 
     assert cursor.get_stream_state()[cursor_field] == "01-03-2023"
 
@@ -478,7 +477,7 @@ def test_given_different_format_and_slice_is_highest_when_close_slice_then_state
     record_cursor_value = "2023-01-03"
     record = Record({cursor_field: record_cursor_value}, _slice)
     cursor.observe(_slice, record)
-    cursor.close_slice(_slice, record)
+    cursor.close_slice(_slice)
 
     assert cursor.get_stream_state()[cursor_field] == "2023-01-03"
 

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

@@ -212,12 +212,11 @@ def test_close_slice(mocked_cursor_factory, mocked_partition_router):
     stream_slice = StreamSlice(partition={"partition key": "first partition"}, cursor_slice={})
     mocked_partition_router.stream_slices.return_value = [stream_slice]
     cursor = PerPartitionCursor(mocked_cursor_factory, mocked_partition_router)
-    last_record = Mock()
     list(cursor.stream_slices())  # generate internal state
 
-    cursor.close_slice(stream_slice, last_record)
+    cursor.close_slice(stream_slice)
 
-    underlying_cursor.close_slice.assert_called_once_with(stream_slice.cursor_slice, Record(last_record.data, stream_slice.cursor_slice))
+    underlying_cursor.close_slice.assert_called_once_with(stream_slice.cursor_slice)
 
 
 def test_given_no_last_record_when_close_slice_then_do_not_raise_error(mocked_cursor_factory, mocked_partition_router):
@@ -228,9 +227,9 @@ def test_given_no_last_record_when_close_slice_then_do_not_raise_error(mocked_cu
     cursor = PerPartitionCursor(mocked_cursor_factory, mocked_partition_router)
     list(cursor.stream_slices())  # generate internal state
 
-    cursor.close_slice(stream_slice, None)
+    cursor.close_slice(stream_slice)
 
-    underlying_cursor.close_slice.assert_called_once_with(stream_slice.cursor_slice, None)
+    underlying_cursor.close_slice.assert_called_once_with(stream_slice.cursor_slice)
 
 
 def test_given_unknown_partition_when_close_slice_then_raise_error():
@@ -239,7 +238,7 @@ def test_given_unknown_partition_when_close_slice_then_raise_error():
     cursor = PerPartitionCursor(any_cursor_factory, any_partition_router)
     stream_slice = StreamSlice(partition={"unknown_partition": "unknown"}, cursor_slice={})
     with pytest.raises(ValueError):
-        cursor.close_slice(stream_slice, Record({}, stream_slice))
+        cursor.close_slice(stream_slice)
 
 
 def test_given_unknown_partition_when_should_be_synced_then_raise_error():

airbyte-cdk/python/unit_tests/sources/declarative/retrievers/test_simple_retriever.py

@@ -445,7 +445,7 @@ def retriever_read_pages(_, __, ___):
         side_effect=retriever_read_pages,
     ):
         list(retriever.read_records(stream_slice=stream_slice, records_schema={}))
-        cursor.close_slice.assert_called_once_with(stream_slice, first_record if first_greater_than_second else second_record)
+        cursor.close_slice.assert_called_once_with(stream_slice)
 
 
 def test_given_stream_data_is_not_record_when_read_records_then_update_slice_with_optional_record():
@@ -478,7 +478,7 @@ def retriever_read_pages(_, __, ___):
     ):
         list(retriever.read_records(stream_slice=stream_slice, records_schema={}))
         cursor.observe.assert_not_called()
-        cursor.close_slice.assert_called_once_with(stream_slice, None)
+        cursor.close_slice.assert_called_once_with(stream_slice)
 
 
 def _generate_slices(number_of_slices):