feat: Add get_record() method to fetch individual records by primary key

- Add LookupValue type alias for primary key values
- Add get_record() method to Source class (keyword-only pk_value arg)
- Add helper methods _get_stream_primary_key() and _normalize_and_validate_pk()
- Add fetch_record() method to DeclarativeExecutor
- Support both direct PK values and dict input with validation
- Dict validation ensures single entry and key matches stream's primary key
- Only supported for declarative (YAML-based) sources
- Add comprehensive unit tests (16 tests, all passing)

This builds on airbytehq/airbyte-python-cdk#846

Co-Authored-By: AJ Steers <[email protected]>

Author: devin-ai-integration[bot]

airbyte/_executors/declarative.py

@@ -140,3 +140,34 @@ def install(self) -> None:
     def uninstall(self) -> None:
         """No-op. The declarative source is included with PyAirbyte."""
         pass
+
+    def fetch_record(
+        self,
+        stream_name: str,
+        pk_value: str,
+        config: dict[str, Any] | None = None,
+    ) -> dict[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 as a string
+            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
+            NotImplementedError: If the stream doesn't support fetching individual records
+            RecordNotFoundException: If the record is not found
+        """
+        merged_config = {**self._config_dict}
+        if config:
+            merged_config.update(config)
+        
+        return self.declarative_source.fetch_record(
+            stream_name=stream_name,
+            pk_value=pk_value,
+            config=merged_config,
+        )

airbyte/sources/base.py

@@ -64,6 +64,9 @@
 )
 
 
+LookupValue = Any
+
+
 class Source(ConnectorBase):  # noqa: PLR0904
     """A class representing a source that can be called."""
 
@@ -577,6 +580,154 @@ def _with_logging(records: Iterable[dict[str, Any]]) -> Iterator[dict[str, Any]]
             progress_tracker=progress_tracker,
         )
 
+    def _get_stream_primary_key(self, stream_name: str) -> list[str]:
+        """Get the primary key for a stream.
+        
+        Returns the primary key from the configured catalog, which includes user overrides.
+        The Airbyte protocol represents primary keys as a list of field paths (list of lists).
+        This method flattens single-field paths to simple strings.
+        
+        Args:
+            stream_name: Name of the stream
+            
+        Returns:
+            List of primary key field names (flattened from field paths)
+            
+        Raises:
+            PyAirbyteInputError: If the stream doesn't exist
+        """
+        configured_catalog = self.get_configured_catalog(streams=[stream_name])
+        
+        if len(configured_catalog.streams) == 0:
+            raise exc.PyAirbyteInputError(
+                message="Stream name does not exist in catalog.",
+                input_value=stream_name,
+            )
+        
+        configured_stream = configured_catalog.streams[0]
+        pk_field_paths = configured_stream.primary_key or []
+        
+        return [field_path[0] for field_path in pk_field_paths if field_path]
+
+    def _normalize_and_validate_pk(
+        self,
+        stream_name: str,
+        pk_value: LookupValue | dict[str, LookupValue],
+    ) -> str:
+        """Normalize and validate primary key input.
+        
+        Accepts either a direct primary key value or a dict with a single entry
+        where the key matches the stream's primary key field name.
+        
+        Args:
+            stream_name: Name of the stream
+            pk_value: Either a direct PK value or a dict with single entry
+            
+        Returns:
+            The primary key value as a string
+            
+        Raises:
+            PyAirbyteInputError: If validation fails
+            NotImplementedError: If the stream has composite or no primary key
+        """
+        primary_key = self._get_stream_primary_key(stream_name)
+        
+        if len(primary_key) == 0:
+            raise NotImplementedError(
+                f"Stream '{stream_name}' does not have a primary key defined. "
+                "Cannot fetch individual records without a primary key."
+            )
+        
+        if len(primary_key) > 1:
+            raise NotImplementedError(
+                f"Stream '{stream_name}' has a composite primary key {primary_key}. "
+                "Fetching by composite primary key is not yet supported."
+            )
+        
+        pk_field = primary_key[0]
+        
+        if isinstance(pk_value, dict):
+            if len(pk_value) != 1:
+                raise exc.PyAirbyteInputError(
+                    message="When passing a dict for pk_value, it must contain exactly one entry.",
+                    input_value=pk_value,
+                    context={
+                        "stream_name": stream_name,
+                        "expected_entries": 1,
+                        "actual_entries": len(pk_value),
+                    },
+                )
+            
+            dict_key = next(iter(pk_value.keys()))
+            dict_value = pk_value[dict_key]
+            
+            if dict_key != pk_field:
+                raise exc.PyAirbyteInputError(
+                    message="The key in the pk_value dict does not match the stream's primary key.",
+                    input_value=dict_key,
+                    context={
+                        "stream_name": stream_name,
+                        "expected_key": pk_field,
+                        "actual_key": dict_key,
+                    },
+                )
+            
+            return str(dict_value)
+        
+        return str(pk_value)
+
+    def get_record(
+        self,
+        stream_name: str,
+        *,
+        pk_value: LookupValue | dict[str, LookupValue],
+    ) -> dict[str, Any]:
+        """Fetch a single record from a stream by primary key.
+        
+        This method is only supported for declarative (YAML-based) sources.
+        
+        Args:
+            stream_name: Name of the stream to fetch from
+            pk_value: Either a direct primary key value (e.g., "123") or a dict
+                with a single entry where the key is the primary key field name
+                and the value is the primary key value (e.g., {"id": "123"}).
+                The dict form provides explicit validation that you're using
+                the correct primary key field.
+                
+        Returns:
+            The fetched record as a dict
+            
+        Raises:
+            NotImplementedError: If the source is not a declarative source, or if
+                the stream has a composite primary key or no primary key
+            PyAirbyteInputError: If the stream doesn't exist or pk_value validation fails
+            RecordNotFoundException: If the record is not found (from CDK)
+            
+        Example:
+            ```python
+            source = get_source("source-rest-api-tutorial", config=config)
+            
+            record = source.get_record("users", pk_value="123")
+            
+            record = source.get_record("users", pk_value={"id": "123"})
+            ```
+        """
+        from airbyte._executors.declarative import DeclarativeExecutor
+        
+        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__}."
+            )
+        
+        validated_pk_str = self._normalize_and_validate_pk(stream_name, pk_value)
+        
+        return self.executor.fetch_record(
+            stream_name=stream_name,
+            pk_value=validated_pk_str,
+            config=self._config_dict,
+        )
+
     def get_documents(
         self,
         stream: str,