feat: Add Source.fetch_record() andStream.fetch_record() backed by SimpleRetriever._fetch_one()

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -77,6 +77,7 @@
     ModelToComponentFactory,
 )
 from airbyte_cdk.sources.declarative.resolvers import COMPONENTS_RESOLVER_TYPE_MAPPING
+from airbyte_cdk.sources.declarative.retrievers import SimpleRetriever
 from airbyte_cdk.sources.declarative.spec.spec import Spec
 from airbyte_cdk.sources.declarative.types import Config, ConnectionDefinition
 from airbyte_cdk.sources.message.concurrent_repository import ConcurrentMessageRepository
@@ -507,6 +508,62 @@ def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCon
             return AirbyteConnectionStatus(status=Status.FAILED, message=repr(error))
         return AirbyteConnectionStatus(status=Status.SUCCEEDED)
 
+    def fetch_record(
+        self,
+        stream_name: str,
+        pk_value: str,
+        config: Optional[Mapping[str, Any]] = None,
+    ) -> Mapping[str, Any]:
+        """
+        Fetch a single record from a stream by primary key.
+
+        Args:
+            stream_name: Name of the stream to fetch from
+            pk_value: Primary key value to fetch as a string (e.g., "123")
+            config: Source configuration (optional, uses instance config if not provided)
+
+        Returns:
+            The fetched record as a dict
+
+        Raises:
+            ValueError: If the stream name is not found in the source
+            NotImplementedError: If the stream doesn't support fetching individual records
+            RecordNotFoundException: If the record is not found
+        """
+        config = config or self._config
+
+        stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
+
+        stream_config = None
+        for config_item in stream_configs:
+            if config_item.get("name") == stream_name:
+                stream_config = config_item
+                break
+
+        if not stream_config:
+            available_streams = [c.get("name") for c in stream_configs]
+            raise ValueError(
+                f"Stream '{stream_name}' not found in source. "
+                f"Available streams: {', '.join(available_streams)}"
+            )
+
+        declarative_stream = self._constructor.create_component(
+            DeclarativeStreamModel,
+            stream_config,
+            config,
+            emit_connector_builder_messages=self._emit_connector_builder_messages,
+        )
+
+        if not isinstance(declarative_stream.retriever, SimpleRetriever):
+            raise NotImplementedError(
+                f"Stream '{stream_name}' does not support fetching individual records. "
+                "Only streams with SimpleRetriever currently support this operation."
+            )
+
+        return declarative_stream.retriever.fetch_one(
+            pk_value=pk_value, records_schema=declarative_stream.get_json_schema()
+        )
+
     @property
     def dynamic_streams(self) -> List[Dict[str, Any]]:
         return self._dynamic_stream_configs(

airbyte_cdk/sources/declarative/exceptions.py

@@ -7,3 +7,7 @@ class ReadException(Exception):
     """
     Raise when there is an error reading data from an API Source
     """
+
+
+class RecordNotFoundException(ReadException):
+    """Raised when a requested record is not found (e.g., 404 response)."""

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -26,6 +26,7 @@
 from airbyte_cdk.legacy.sources.declarative.incremental import ResumableFullRefreshCursor
 from airbyte_cdk.legacy.sources.declarative.incremental.declarative_cursor import DeclarativeCursor
 from airbyte_cdk.models import AirbyteMessage
+from airbyte_cdk.sources.declarative.exceptions import RecordNotFoundException
 from airbyte_cdk.sources.declarative.extractors.http_selector import HttpSelector
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.partition_routers.single_partition_router import (
@@ -626,6 +627,109 @@ def _to_partition_key(to_serialize: Any) -> str:
         # separators have changed in Python 3.4. To avoid being impacted by further change, we explicitly specify our own value
         return json.dumps(to_serialize, indent=None, separators=(",", ":"), sort_keys=True)
 
+    def fetch_one(
+        self,
+        pk_value: str,
+        records_schema: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        """Fetch a single record by primary key value.
+
+        This method constructs a path by appending the primary key value to the base path
+        and sends a GET request to fetch a single record. It's designed for REST APIs that
+        follow the convention: GET /resource/{id}
+
+        Args:
+            pk_value: The primary key value to fetch as a string (e.g., "123")
+            records_schema: JSON schema describing the record structure
+
+        Returns:
+            The fetched record as a dict.
+
+        Raises:
+            RecordNotFoundException: If the response is empty/ignored or parsing yields no records.
+            Exception: HTTP errors (including 404) are propagated from requester's error handling.
+
+        Example:
+            record = retriever.fetch_one("123", schema)
+
+        Note:
+            This implementation uses convention-based path construction, appending /{pk_value} to the base path. (important-comment)
+
+            Alternative approaches that could be implemented in the future: (important-comment)
+            - Path template interpolation: Use a configurable template like "{base_path}/{id}" (important-comment)
+              See: https://github.com/airbytehq/airbyte-python-cdk/issues/833#phase-1a (important-comment)
+            - Field path configuration: Allow specifying which response field contains the record (important-comment)
+              for APIs that wrap single records in envelopes like {"data": {...}} (important-comment)
+        """
+        # Single-record fetch doesn't involve partitioning, so we pass an empty StreamSlice
+        empty_stream_slice = StreamSlice(partition={}, cursor_slice={})
+
+        # Get the base path from the requester
+        base_path = self.requester.get_path(
+            stream_state={},
+            stream_slice=empty_stream_slice,
+            next_page_token=None,
+        )
+
+        fetch_path = f"{base_path.rstrip('/')}/{pk_value.lstrip('/')}"
+
+        # send_request() may return None when the error handler chooses to IGNORE a response
+        response: requests.Response | None = self.requester.send_request(
+            path=fetch_path,
+            stream_state={},
+            stream_slice=empty_stream_slice,
+            next_page_token=None,
+            request_headers=self._request_headers(
+                stream_state={},
+                stream_slice=empty_stream_slice,
+                next_page_token=None,
+            ),
+            request_params=self._request_params(
+                stream_state={},
+                stream_slice=empty_stream_slice,
+                next_page_token=None,
+            ),
+            request_body_data=self._request_body_data(
+                stream_state={},
+                stream_slice=empty_stream_slice,
+                next_page_token=None,
+            ),
+            request_body_json=self._request_body_json(
+                stream_state={},
+                stream_slice=empty_stream_slice,
+                next_page_token=None,
+            ),
+            log_formatter=self.log_formatter,
+        )
+
+        if not response:
+            raise RecordNotFoundException(
+                f"Record with primary key {pk_value} not found (no response)"
+            )
+
+        records_iter: Iterable[Record] = self._parse_response(
+            response=response,
+            stream_state={},
+            records_schema=records_schema,
+            stream_slice=empty_stream_slice,
+            next_page_token=None,
+        )
+
+        first_record: Record | None = next(iter(records_iter), None)
+        if first_record:
+            return dict(first_record.data)
+
+        try:
+            response_body = response.json()
+            if isinstance(response_body, dict) and response_body:
+                return response_body
+        except Exception:
+            pass
+
+        raise RecordNotFoundException(
+            f"Record with primary key {pk_value} not found (empty response)"
+        )
+
 
 def _deep_merge(
     target: MutableMapping[str, Any], source: Union[Record, MutableMapping[str, Any]]

airbyte_cdk/sources/streams/concurrent/abstract_stream.py

@@ -90,3 +90,22 @@ def check_availability(self) -> StreamAvailability:
         """
         :return: If the stream is available and if not, why
         """
+
+    def fetch_record(self, pk_value: str) -> Mapping[str, Any]:
+        """
+        Fetch a single record by primary key value.
+
+        Args:
+            pk_value: The primary key value as a string (e.g., "123")
+
+        Returns:
+            The fetched record as a dict
+
+        Raises:
+            NotImplementedError: If the stream doesn't support fetching individual records
+            RecordNotFoundException: If the record is not found
+        """
+        raise NotImplementedError(
+            f"Stream {self.name} does not support fetching individual records. "
+            "Only declarative streams with SimpleRetriever currently support this operation."
+        )