Repository: meltano/sdk
Pull request #2765 (2024-11-14 03:43:10)

feat(targets): The last received Singer message is now logged when the target fails


<!-- readthedocs-preview meltano-sdk start -->
----
📚 Documentation preview 📚: https://meltano-sdk--2765.org.readthedocs.build/en/2765/

<!-- readthedocs-preview meltano-sdk end -->

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/meltano/sdk/pull/2765?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 90.81%. Comparing base [(`25527c4`)](https://app.codecov.io/gh/meltano/sdk/commit/25527c4a18785865d1552ceb1e9be0f00539eb80?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) to head [(`7ad9f5e`)](https://app.codecov.io/gh/meltano/sdk/commit/7ad9f5e83aab1ae6ab3160cbd7f602035fb3bf38?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #2765      +/-   ##
==========================================
+ Coverage   90.80%   90.81%   +0.01%     
==========================================
  Files          62       62              
  Lines        5122     5130       +8     
  Branches      660      660              
==========================================
+ Hits         4651     4659       +8     
  Misses        330      330              
  Partials      141      141              
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/meltano/sdk/pull/2765?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).


Patch:
diff --git a/singer_sdk/_singerlib/encoding/_base.py b/singer_sdk/_singerlib/encoding/_base.py
index 62793776..404c6ed6 100644
--- a/singer_sdk/_singerlib/encoding/_base.py
+++ b/singer_sdk/_singerlib/encoding/_base.py
@@ -38,6 +38,10 @@ class SingerMessageType(str, enum.Enum):
 class GenericSingerReader(t.Generic[T], metaclass=abc.ABCMeta):
     """Interface for all plugins reading Singer messages as strings or bytes."""
 
+    def __init__(self) -> None:
+        super().__init__()
+        self._current_message: T | None = None
+
     @t.final
     def listen(self, file_input: t.IO[T] | None = None) -> None:
         """Read from input until all messages are processed.
@@ -45,7 +49,14 @@ class GenericSingerReader(t.Generic[T], metaclass=abc.ABCMeta):
         Args:
             file_input: Readable stream of messages. Defaults to standard in.
         """
-        self._process_lines(file_input or self.default_input)
+        try:
+            self._process_lines(file_input or self.default_input)
+        except Exception:
+            logger.debug(
+                "Failed while processing Singer message: %s",
+                self._current_message,
+            )
+            raise
         self._process_endofpipe()
 
     def _process_lines(self, file_input: t.IO[T]) -> t.Counter[str]:
@@ -59,6 +70,8 @@ class GenericSingerReader(t.Generic[T], metaclass=abc.ABCMeta):
         """
         stats: dict[str, int] = defaultdict(int)
         for line in file_input:
+            self._current_message = line
+
             line_dict = self.deserialize_json(line)
             self._assert_line_requires(line_dict, requires={"type"})
 


Test patch:
diff --git a/tests/core/test_io.py b/tests/core/test_io.py
index a48a785d..4f8b2019 100644
--- a/tests/core/test_io.py
+++ b/tests/core/test_io.py
@@ -6,6 +6,7 @@ import decimal
 import io
 import itertools
 import json
+import logging
 from contextlib import nullcontext, redirect_stdout
 from textwrap import dedent
 
@@ -83,6 +84,29 @@ def test_listen_unknown_message():
         reader.listen(input_lines)
 
 
+def test_listen_error(caplog: pytest.LogCaptureFixture):
+    class ErrorReader(DummyReader):
+        def _process_record_message(self, message_dict: dict) -> None:  # noqa: ARG002
+            msg = "Bad record"
+            raise ValueError(msg)
+
+    message = RecordMessage(
+        stream="users",
+        record={"id": 1, "value": 1.23},
+    )
+
+    input_lines = io.StringIO(json.dumps(message.to_dict()) + "\n")
+
+    reader = ErrorReader()
+    with caplog.at_level(logging.DEBUG), pytest.raises(ValueError, match="Bad record"):
+        reader.listen(input_lines)
+
+    assert caplog.records
+
+    message = caplog.records[0].message
+    assert "Failed while processing Singer message" in message
+
+
 def test_write_message():
     writer = SingerWriter()
     message = RecordMessage(