Repository: meltano/sdk
Pull request #1927 (2023-08-27 05:29:17)

fix: Handle replication key not found in stream schema
Closes: #1332

Changes:

- Add new `InvalidReplicationKeyException` to `singer_sdk.exceptions`
- Update `streams.core`. A `get()` is used to obtain the `replication_key` from `self.schema`, if the key is not present in the schema, the variable `type_dict` would equal `None`. A check has been added, if the value of the `type_dict` is `None` then the new exception is raised
- The resulting error message is of the format:

```log
"incorrect_replication_key" is not in schema for stream name: my_stream"
```

- Add a test to `tests/core/test_streams.py` which checks that a `SimpleTestStream` with `replication_key` overridden to `"INVALID"` raises the correct `InvalidReplicationKeyException` exception. 



codecov[bot]: ## [Codecov](https://app.codecov.io/gh/meltano/sdk/pull/1927?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) Report
> Merging [#1927](https://app.codecov.io/gh/meltano/sdk/pull/1927?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) (4ddf1c8) into [main](https://app.codecov.io/gh/meltano/sdk/commit/712ce9a8cac1502390ef0f81376d73264f5f600d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) (712ce9a) will **decrease** coverage by `0.05%`.
> The diff coverage is `25.00%`.

```diff
@@            Coverage Diff             @@
##             main    #1927      +/-   ##
==========================================
- Coverage   86.96%   86.91%   -0.05%     
==========================================
  Files          59       59              
  Lines        5087     5091       +4     
  Branches      824      825       +1     
==========================================
+ Hits         4424     4425       +1     
- Misses        466      468       +2     
- Partials      197      198       +1     
```


| [Files Changed](https://app.codecov.io/gh/meltano/sdk/pull/1927?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) | Coverage Δ | |
|---|---|---|
| [singer\_sdk/streams/core.py](https://app.codecov.io/gh/meltano/sdk/pull/1927?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay9zdHJlYW1zL2NvcmUucHk=) | `83.37% <0.00%> (-0.59%)` | :arrow_down: |
| [singer\_sdk/exceptions.py](https://app.codecov.io/gh/meltano/sdk/pull/1927?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay9leGNlcHRpb25zLnB5) | `100.00% <100.00%> (ø)` | |

:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano)

edgarrmondragon: Thanks @mjsqu!

Patch:
diff --git a/singer_sdk/exceptions.py b/singer_sdk/exceptions.py
index 23325aa2..35177629 100644
--- a/singer_sdk/exceptions.py
+++ b/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."""
 
diff --git a/singer_sdk/streams/core.py b/singer_sdk/streams/core.py
index 53e9533f..4c3adb22 100644
--- a/singer_sdk/streams/core.py
+++ b/singer_sdk/streams/core.py
@@ -19,6 +19,7 @@ from singer_sdk.batch import JSONLinesBatcher
 from singer_sdk.exceptions import (
     AbortedSyncFailedException,
     AbortedSyncPausedException,
+    InvalidReplicationKeyException,
     InvalidStreamSortException,
     MaxRecordsLimitException,
 )
@@ -211,10 +212,17 @@ class Stream(metaclass=abc.ABCMeta):
 
         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(


Test patch:
diff --git a/tests/core/test_streams.py b/tests/core/test_streams.py
index 34bbc751..a3a45108 100644
--- a/tests/core/test_streams.py
+++ b/tests/core/test_streams.py
@@ -10,6 +10,9 @@ import pytest
 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",
     [