feat: Add singleton record fetch functionality to PyAirbyte

Implements Source.get_record() and DeclarativeExecutor.fetch_record() methods
to enable fetching single records by primary key value from declarative sources.

Key features:
- Source.get_record(stream_name, pk_value) - Public API for fetching records
- DeclarativeExecutor.fetch_record() - Internal implementation using CDK components
- Primary key validation and normalization (supports string, int, dict formats)
- Composite primary key detection (raises NotImplementedError)
- New AirbyteRecordNotFoundError exception for missing records
- Comprehensive unit tests with proper mocking

This implementation reuses existing CDK components (SimpleRetriever, HttpClient,
RecordSelector) without monkey-patching or pinning CDK versions, providing a
hybrid approach that works with the current CDK release.

Related to CDK PR airbytehq/airbyte-python-cdk#846

Co-Authored-By: AJ Steers <aj@airbyte.io>

Author: devin-ai-integration[bot]

airbyte/_executors/declarative.py

@@ -15,7 +15,13 @@
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     ConcurrentDeclarativeSource,
 )
+from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import (
+    ModelToComponentFactory,
+)
+from airbyte_cdk.sources.declarative.retrievers.simple_retriever import SimpleRetriever
+from airbyte_cdk.sources.types import StreamSlice
 
+from airbyte import exceptions as exc
 from airbyte._executors.base import Executor
 
 
@@ -140,3 +146,152 @@ def install(self) -> None:
     def uninstall(self) -> None:
         """No-op. The declarative source is included with PyAirbyte."""
         pass
+
+    def fetch_record(  # noqa: PLR0914
+        self,
+        stream_name: str,
+        primary_key_value: str,
+        config: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """Fetch a single record by primary key from a declarative stream.
+
+        This method constructs an HTTP GET request to fetch a single record by appending
+        the primary key value to the stream's base path (e.g., /users/123).
+
+        Args:
+            stream_name: The name of the stream to fetch from.
+            primary_key_value: The primary key value as a string.
+            config: Optional config overrides to merge with the executor's config.
+
+        Returns:
+            The fetched record as a dictionary.
+
+        Raises:
+            exc.AirbyteStreamNotFoundError: If the stream is not found in the manifest.
+            exc.AirbyteRecordNotFoundError: If the record is not found (empty response).
+            NotImplementedError: If the stream does not use SimpleRetriever.
+        """
+        merged_config = {**self._config_dict, **(config or {})}
+
+        stream_configs = self._manifest_dict.get("streams", [])
+        stream_config = None
+        for config_item in stream_configs:
+            if config_item.get("name") == stream_name:
+                stream_config = config_item
+                break
+
+        if stream_config is None:
+            available_streams = [s.get("name") for s in stream_configs]
+            raise exc.AirbyteStreamNotFoundError(
+                stream_name=stream_name,
+                connector_name=self.name,
+                available_streams=available_streams,
+                message=f"Stream '{stream_name}' not found in manifest.",
+            )
+
+        factory = ModelToComponentFactory()
+
+        retriever_config = stream_config.get("retriever")
+        if retriever_config is None:
+            raise NotImplementedError(
+                f"Stream '{stream_name}' does not have a retriever configuration. "
+                "fetch_record() is only supported for streams with retrievers."
+            )
+
+        try:
+            retriever = factory.create_component(
+                model_type=type(retriever_config),  # type: ignore[arg-type]
+                component_definition=retriever_config,
+                config=merged_config,
+            )
+        except Exception as e:
+            raise NotImplementedError(
+                f"Failed to create retriever for stream '{stream_name}': {e}"
+            ) from e
+
+        if not isinstance(retriever, SimpleRetriever):
+            raise NotImplementedError(
+                f"Stream '{stream_name}' uses {type(retriever).__name__}, but fetch_record() "
+                "only supports SimpleRetriever."
+            )
+
+        empty_slice = StreamSlice(partition={}, cursor_slice={})
+        base_path = retriever.requester.get_path(
+            stream_state={},
+            stream_slice=empty_slice,
+            next_page_token=None,
+        )
+
+        fetch_path = f"{base_path}/{primary_key_value}".lstrip("/")
+
+        response = retriever.requester.send_request(
+            path=fetch_path,
+            stream_state={},
+            stream_slice=empty_slice,
+            next_page_token=None,
+            request_headers=retriever._request_headers(  # noqa: SLF001
+                stream_slice=empty_slice,
+                next_page_token=None,
+            ),
+            request_params=retriever._request_params(  # noqa: SLF001
+                stream_slice=empty_slice,
+                next_page_token=None,
+            ),
+            request_body_data=retriever._request_body_data(  # noqa: SLF001
+                stream_slice=empty_slice,
+                next_page_token=None,
+            ),
+            request_body_json=retriever._request_body_json(  # noqa: SLF001
+                stream_slice=empty_slice,
+                next_page_token=None,
+            ),
+        )
+
+        if response is None:
+            msg = (
+                f"No response received when fetching record with primary key "
+                f"'{primary_key_value}' from stream '{stream_name}'."
+            )
+            raise exc.AirbyteRecordNotFoundError(
+                stream_name=stream_name,
+                primary_key_value=primary_key_value,
+                connector_name=self.name,
+                message=msg,
+            )
+
+        schema = stream_config.get("schema_loader", {})
+        records_schema = schema if isinstance(schema, dict) else {}
+
+        records = list(
+            retriever.record_selector.select_records(
+                response=response,
+                stream_state={},
+                records_schema=records_schema,
+                stream_slice=empty_slice,
+                next_page_token=None,
+            )
+        )
+
+        if not records:
+            try:
+                response_json = response.json()
+                if isinstance(response_json, dict) and response_json:
+                    return response_json
+            except Exception:
+                pass
+
+            msg = (
+                f"Record with primary key '{primary_key_value}' "
+                f"not found in stream '{stream_name}'."
+            )
+            raise exc.AirbyteRecordNotFoundError(
+                stream_name=stream_name,
+                primary_key_value=primary_key_value,
+                connector_name=self.name,
+                message=msg,
+            )
+
+        first_record = records[0]
+        if hasattr(first_record, "data"):
+            return dict(first_record.data)  # type: ignore[arg-type]
+        return dict(first_record)  # type: ignore[arg-type]

airbyte/exceptions.py

@@ -412,6 +412,14 @@ class AirbyteStateNotFoundError(AirbyteConnectorError, KeyError):
     available_streams: list[str] | None = None
 
 
+@dataclass
+class AirbyteRecordNotFoundError(AirbyteConnectorError):
+    """Record not found in stream."""
+
+    stream_name: str | None = None
+    primary_key_value: str | None = None
+
+
 @dataclass
 class PyAirbyteSecretNotFoundError(PyAirbyteError):
     """Secret not found."""

airbyte/sources/base.py

@@ -30,6 +30,7 @@
 
 from airbyte import exceptions as exc
 from airbyte._connector_base import ConnectorBase
+from airbyte._executors.declarative import DeclarativeExecutor
 from airbyte._message_iterators import AirbyteMessageIterator
 from airbyte._util.temp_files import as_temp_files
 from airbyte.caches.util import get_default_cache
@@ -601,6 +602,117 @@ def get_documents(
             render_metadata=render_metadata,
         )
 
+    def _get_stream_primary_key(self, stream_name: str) -> list[str]:
+        """Get the primary key for a stream.
+
+        Returns the primary key as a flat list of field names.
+        Handles the Airbyte protocol's nested list structure.
+        """
+        catalog = self.configured_catalog
+        for configured_stream in catalog.streams:
+            if configured_stream.stream.name == stream_name:
+                pk = configured_stream.primary_key
+                if not pk:
+                    return []
+                if isinstance(pk, list) and len(pk) > 0:
+                    if isinstance(pk[0], list):
+                        return [field[0] if isinstance(field, list) else field for field in pk]
+                    return list(pk)  # type: ignore[arg-type]
+                return []
+        raise exc.AirbyteStreamNotFoundError(
+            stream_name=stream_name,
+            connector_name=self.name,
+            available_streams=self.get_available_streams(),
+        )
+
+    def _normalize_and_validate_pk_value(
+        self,
+        stream_name: str,
+        pk_value: Any,  # noqa: ANN401
+    ) -> str:
+        """Normalize and validate a primary key value.
+
+        Accepts:
+        - A string or int (converted to string)
+        - A dict with a single entry matching the stream's primary key field
+
+        Returns the PK value as a string.
+        """
+        primary_key_fields = self._get_stream_primary_key(stream_name)
+
+        if not primary_key_fields:
+            raise exc.PyAirbyteInputError(
+                message=f"Stream '{stream_name}' does not have a primary key defined.",
+                input_value=str(pk_value),
+            )
+
+        if len(primary_key_fields) > 1:
+            raise NotImplementedError(
+                f"Stream '{stream_name}' has a composite primary key {primary_key_fields}. "
+                "Fetching by composite primary key is not yet supported."
+            )
+
+        pk_field = primary_key_fields[0]
+
+        if isinstance(pk_value, dict):
+            if len(pk_value) != 1:
+                raise exc.PyAirbyteInputError(
+                    message="When providing pk_value as a dict, it must contain exactly one entry.",
+                    input_value=str(pk_value),
+                )
+            provided_key = next(iter(pk_value.keys()))
+            if provided_key != pk_field:
+                msg = (
+                    f"Primary key field '{provided_key}' does not match "
+                    f"stream's primary key '{pk_field}'."
+                )
+                raise exc.PyAirbyteInputError(
+                    message=msg,
+                    input_value=str(pk_value),
+                )
+            return str(pk_value[provided_key])
+
+        return str(pk_value)
+
+    def get_record(
+        self,
+        stream_name: str,
+        *,
+        pk_value: Any,  # noqa: ANN401
+    ) -> dict[str, Any]:
+        """Fetch a single record by primary key value.
+
+        This method is currently only supported for declarative (YAML-based) sources.
+
+        Args:
+            stream_name: The name of the stream to fetch from.
+            pk_value: The primary key value. Can be:
+                - A string or integer value (e.g., "123" or 123)
+                - A dict with a single entry (e.g., {"id": "123"})
+
+        Returns:
+            The fetched record as a dictionary.
+
+        Raises:
+            exc.AirbyteStreamNotFoundError: If the stream does not exist.
+            exc.AirbyteRecordNotFoundError: If the record is not found.
+            exc.PyAirbyteInputError: If the pk_value format is invalid.
+            NotImplementedError: If the source is not declarative or uses composite keys.
+        """
+        if not isinstance(self.executor, DeclarativeExecutor):
+            raise NotImplementedError(
+                f"get_record() is only supported for declarative sources. "
+                f"This source uses {type(self.executor).__name__}."
+            )
+
+        pk_value_str = self._normalize_and_validate_pk_value(stream_name, pk_value)
+
+        return self.executor.fetch_record(
+            stream_name=stream_name,
+            primary_key_value=pk_value_str,
+            config=self._config_dict,
+        )
+
     def get_samples(
         self,
         streams: list[str] | Literal["*"] | None = None,