fix: Handle replication key not found in stream schema (#1927)

* Handle replication key not found in stream schema

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Add missing Raises: to is_timestamp_replication_key docstring

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Split raise docstring over to new line for is_timestamp_replication_key docstring

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Correct indentation for raises docstring

* Add test_stream_invalid_replication_key

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Import InvalidReplicationKeyException name

* pre-commit check updates - line length and 'useless' expression

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Rename 'unused' variable with underscore prefix - variable is used to intentionally raise and exception

* Improve error message

Co-authored-by: Edgar R. M. <edgarrm358@gmail.com>

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Update exception message in invalid replication key test

Co-authored-by: Edgar R. M. <edgarrm358@gmail.com>

* Make linter happy

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Edgar R. M <edgar@meltano.com>
Co-authored-by: Edgar R. M. <edgarrm358@gmail.com>

Author: 3 people

singer_sdk/exceptions.py

@@ -17,6 +17,10 @@ class FatalAPIError(Exception):
     """Exception raised when a failed request should not be considered retriable."""
 
 
+class InvalidReplicationKeyException(Exception):
+    """Exception to raise if the replication key is not in the stream properties."""
+
+
 class InvalidStreamSortException(Exception):
     """Exception to raise if sorting errors are found while syncing the records."""
 

singer_sdk/streams/core.py

@@ -19,6 +19,7 @@
 from singer_sdk.exceptions import (
     AbortedSyncFailedException,
     AbortedSyncPausedException,
+    InvalidReplicationKeyException,
     InvalidStreamSortException,
     MaxRecordsLimitException,
 )
@@ -211,10 +212,17 @@ def is_timestamp_replication_key(self) -> bool:
 
         Returns:
             True if the stream uses a timestamp-based replication key.
+
+        Raises:
+            InvalidReplicationKeyException: If the schema does not contain the
+                replication key.
         """
         if not self.replication_key:
             return False
         type_dict = self.schema.get("properties", {}).get(self.replication_key)
+        if type_dict is None:
+            msg = f"Field '{self.replication_key}' is not in schema for stream '{self.name}'"  # noqa: E501
+            raise InvalidReplicationKeyException(msg)
         return is_datetime_type(type_dict)
 
     def get_starting_replication_key_value(

tests/core/test_streams.py

@@ -10,6 +10,9 @@
 import requests
 
 from singer_sdk._singerlib import Catalog, MetadataMapping
+from singer_sdk.exceptions import (
+    InvalidReplicationKeyException,
+)
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.helpers.jsonpath import _compile_jsonpath, extract_jsonpath
 from singer_sdk.pagination import first
@@ -275,6 +278,24 @@ def test_stream_starting_timestamp(
     assert get_starting_value(None) == expected_starting_value
 
 
+def test_stream_invalid_replication_key(tap: SimpleTestTap):
+    """Validate an exception is raised if replication_key not in schema."""
+
+    class InvalidReplicationKeyStream(SimpleTestStream):
+        replication_key = "INVALID"
+
+    stream = InvalidReplicationKeyStream(tap)
+
+    with pytest.raises(
+        InvalidReplicationKeyException,
+        match=(
+            f"Field '{stream.replication_key}' is not in schema for stream "
+            f"'{stream.name}'"
+        ),
+    ):
+        _check = stream.is_timestamp_replication_key
+
+
 @pytest.mark.parametrize(
     "path,content,result",
     [