remove stream_slices() from SimpleRetriever and AsyncRetriever

Author: brianjlai

airbyte_cdk/legacy/sources/declarative/declarative_stream.py

@@ -5,6 +5,8 @@
 from dataclasses import InitVar, dataclass, field
 from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Union
 
+from typing_extensions import deprecated
+
 from airbyte_cdk.legacy.sources.declarative.incremental import (
     GlobalSubstreamCursor,
     PerPartitionCursor,
@@ -13,7 +15,6 @@
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.migrations.state_migration import StateMigration
-from airbyte_cdk.sources.declarative.retrievers import SimpleRetriever
 from airbyte_cdk.sources.declarative.retrievers.async_retriever import AsyncRetriever
 from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
 from airbyte_cdk.sources.declarative.schema import DefaultSchemaLoader
@@ -28,6 +29,7 @@
 from airbyte_cdk.sources.types import Config, StreamSlice
 
 
+@deprecated("DeclarativeStream has been deprecated in favor of the concurrent DefaultStream")
 @dataclass
 class DeclarativeStream(Stream):
     """
@@ -198,8 +200,6 @@ def state_checkpoint_interval(self) -> Optional[int]:
         return None
 
     def get_cursor(self) -> Optional[Cursor]:
-        if self.retriever and isinstance(self.retriever, SimpleRetriever):
-            return self.retriever.cursor
         return None
 
     def _get_checkpoint_reader(

airbyte_cdk/sources/declarative/retrievers/async_retriever.py

@@ -11,7 +11,7 @@
 )
 from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
 from airbyte_cdk.sources.streams.core import StreamData
-from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
+from airbyte_cdk.sources.types import Config, StreamSlice
 from airbyte_cdk.sources.utils.slice_logger import AlwaysLogSliceLogger
 
 
@@ -59,30 +59,6 @@ def exit_on_rate_limit(self, value: bool) -> None:
         if job_orchestrator is not None:
             job_orchestrator._job_repository.creation_requester.exit_on_rate_limit = value  # type: ignore[attr-defined, assignment]
 
-    @property
-    def state(self) -> StreamState:
-        """
-        As a first iteration for sendgrid, there is no state to be managed
-        """
-        return {}
-
-    @state.setter
-    def state(self, value: StreamState) -> None:
-        """
-        As a first iteration for sendgrid, there is no state to be managed
-        """
-        pass
-
-    def _get_stream_state(self) -> StreamState:
-        """
-        Gets the current state of the stream.
-
-        Returns:
-            StreamState: Mapping[str, Any]
-        """
-
-        return self.state
-
     def _validate_and_get_stream_slice_jobs(
         self, stream_slice: Optional[StreamSlice] = None
     ) -> Iterable[AsyncJob]:
@@ -101,9 +77,6 @@ def _validate_and_get_stream_slice_jobs(
         """
         return stream_slice.extra_fields.get("jobs", []) if stream_slice else []
 
-    def stream_slices(self) -> Iterable[Optional[StreamSlice]]:
-        yield from self.stream_slicer.stream_slices()
-
     def read_records(
         self,
         records_schema: Mapping[str, Any],
@@ -112,13 +85,12 @@ def read_records(
         # emit the slice_descriptor log message, for connector builder TestRead
         yield self.slice_logger.create_slice_log_message(stream_slice.cursor_slice)  # type: ignore
 
-        stream_state: StreamState = self._get_stream_state()
         jobs: Iterable[AsyncJob] = self._validate_and_get_stream_slice_jobs(stream_slice)
         records: Iterable[Mapping[str, Any]] = self.stream_slicer.fetch_records(jobs)
 
         yield from self.record_selector.filter_and_transform(
             all_data=records,
-            stream_state=stream_state,
+            stream_state={},  # stream_state as an interpolation context is deprecated
             records_schema=records_schema,
             stream_slice=stream_slice,
         )

airbyte_cdk/sources/declarative/retrievers/retriever.py

@@ -33,7 +33,10 @@ def read_records(
     @abstractmethod
     @deprecated("Stream slicing is being moved to the stream level.")
     def stream_slices(self) -> Iterable[Optional[StreamSlice]]:
-        """Returns the stream slices"""
+        """Does nothing as this method is deprecated, so underlying Retriever implementations
+        do not need to implement this.
+        """
+        yield from []
 
     @property
     @deprecated("State management is being moved to the stream level.")

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -456,23 +456,6 @@ def read_records(
         )
         yield from self._read_pages(record_generator, _slice)
 
-    def stream_slices(self) -> Iterable[Optional[StreamSlice]]:  # type: ignore
-        """
-        Specifies the slices for this stream. See the stream slicing section of the docs for more information.
-
-        :param sync_mode:
-        :param cursor_field:
-        :return:
-        """
-        return self.stream_slicer.stream_slices()
-
-    # todo: There are a number of things that can be cleaned up when we remove self.cursor and all the related
-    #  SimpleRetriever state management that is handled by the concurrent CDK Framework:
-    #  - DONE ModelToComponentFactory.create_datetime_based_cursor() should be removed since it does need to be instantiated
-    #  - DONE ModelToComponentFactory.create_incrementing_count_cursor() should be removed since it's a placeholder
-    #  - DONE test_simple_retriever.py: Remove all imports and usages of legacy cursor components
-    #  - DONE test_model_to_component_factory.py:test_datetime_based_cursor() test can be removed
-
     def _parse_records(
         self,
         response: Optional[requests.Response],

unit_tests/sources/declarative/retrievers/test_simple_retriever.py

@@ -21,10 +21,6 @@
 from airbyte_cdk.sources.declarative.decoders import JsonDecoder
 from airbyte_cdk.sources.declarative.extractors import DpathExtractor, HttpSelector, RecordSelector
 from airbyte_cdk.sources.declarative.partition_routers import SinglePartitionRouter
-from airbyte_cdk.sources.declarative.partition_routers.substream_partition_router import (
-    ParentStreamConfig,
-    SubstreamPartitionRouter,
-)
 from airbyte_cdk.sources.declarative.requesters.paginators import DefaultPaginator, Paginator
 from airbyte_cdk.sources.declarative.requesters.paginators.strategies import (
     CursorPaginationStrategy,
@@ -42,7 +38,6 @@
 from airbyte_cdk.sources.declarative.requesters.requester import HttpMethod, Requester
 from airbyte_cdk.sources.declarative.retrievers.pagination_tracker import PaginationTracker
 from airbyte_cdk.sources.declarative.retrievers.simple_retriever import SimpleRetriever
-from airbyte_cdk.sources.declarative.stream_slicers import StreamSlicerTestReadDecorator
 from airbyte_cdk.sources.streams.http.pagination_reset_exception import (
     PaginationResetRequiredException,
 )
@@ -425,30 +420,6 @@ def test_path(test_name, requester_path, paginator_path, expected_path):
     assert actual_path == expected_path
 
 
-def test_limit_stream_slices():
-    maximum_number_of_slices = 4
-    stream_slicer = MagicMock()
-    stream_slicer.stream_slices.return_value = _generate_slices(maximum_number_of_slices * 2)
-    stream_slicer_wrapped = StreamSlicerTestReadDecorator(
-        wrapped_slicer=stream_slicer,
-        maximum_number_of_slices=maximum_number_of_slices,
-    )
-    retriever = SimpleRetriever(
-        name="stream_name",
-        primary_key=primary_key,
-        requester=MagicMock(),
-        paginator=MagicMock(),
-        record_selector=MagicMock(),
-        stream_slicer=stream_slicer_wrapped,
-        parameters={},
-        config={},
-    )
-
-    truncated_slices = list(retriever.stream_slices())
-
-    assert truncated_slices == _generate_slices(maximum_number_of_slices)
-
-
 def test_given_stream_data_is_not_record_when_read_records_then_update_slice_with_optional_record():
     stream_data = [
         AirbyteMessage(