yield slice cursor info for the TestRead

Author: Oleksandr Bazarnov

airbyte_cdk/sources/declarative/retrievers/async_retriever.py

@@ -1,13 +1,12 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 
 
-from dataclasses import InitVar, dataclass
+from dataclasses import InitVar, dataclass, field
 from typing import Any, Iterable, Mapping, Optional
 
 from typing_extensions import deprecated
 
 from airbyte_cdk.sources.declarative.async_job.job import AsyncJob
-from airbyte_cdk.sources.declarative.async_job.job_orchestrator import AsyncPartition
 from airbyte_cdk.sources.declarative.extractors.record_selector import RecordSelector
 from airbyte_cdk.sources.declarative.partition_routers.async_job_partition_router import (
     AsyncJobPartitionRouter,
@@ -16,6 +15,7 @@
 from airbyte_cdk.sources.source import ExperimentalClassWarning
 from airbyte_cdk.sources.streams.core import StreamData
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
+from airbyte_cdk.sources.utils.slice_logger import AlwaysLogSliceLogger
 
 
 @deprecated(
@@ -28,6 +28,10 @@ class AsyncRetriever(Retriever):
     parameters: InitVar[Mapping[str, Any]]
     record_selector: RecordSelector
     stream_slicer: AsyncJobPartitionRouter
+    slice_logger: AlwaysLogSliceLogger = field(
+        init=False,
+        default_factory=lambda: AlwaysLogSliceLogger(),
+    )
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._parameters = parameters
@@ -74,14 +78,36 @@ def _validate_and_get_stream_slice_jobs(
         """
         return stream_slice.extra_fields.get("jobs", []) if stream_slice else []
 
+    def _get_cursor_slice_info(
+        self,
+        stream_slice: Optional[StreamSlice] = None,
+    ) -> Mapping[str, Any]:
+        """
+        Retrieve the cursor slice information from a provided stream slice.
+
+        This method checks if a stream slice is provided and contains a 'cursor_slice' attribute.
+        If present, it returns the value of 'cursor_slice' as a dictionary. Otherwise, it returns an empty dictionary.
+
+        Args:
+            stream_slice (Optional[StreamSlice]): An optional stream slice object that may include a 'cursor_slice' attribute.
+
+        Returns:
+            Mapping[str, Any]: A dictionary containing the cursor slice information if available, otherwise an empty dictionary.
+        """
+        return stream_slice.cursor_slice if stream_slice and stream_slice.cursor_slice else {}
+
     def stream_slices(self) -> Iterable[Optional[StreamSlice]]:
-        return self.stream_slicer.stream_slices()
+        yield from self.stream_slicer.stream_slices()
 
     def read_records(
         self,
         records_schema: Mapping[str, Any],
         stream_slice: Optional[StreamSlice] = None,
     ) -> Iterable[StreamData]:
+        # emit the slice_descriptor log message, for connector builder TestRead
+        yield self.slice_logger.create_slice_log_message(
+            self._get_cursor_slice_info(stream_slice),
+        )
         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)