fix: Handle replication key not found in stream schema

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",
     [