Repository: meltano/sdk
Pull request #2058 (2023-11-20 17:52:13)

refactor: Implement a `SingerWriter` class in `singer_sdk.io_base` and use it to emit Singer messages
Creates a `SingerWriter` class in `singer_sdk.io_base` which the write_message(), and format_message functions from singer_sdk._singerlib.messages are placed into. After doinng this all the stdin and stdout methods, json encoders, and json decoder would be in one file.  SingerWriter was added as an inherited class of Tap just like SingerReader is an inherited class of Target.

<!-- readthedocs-preview meltano-sdk start -->
----
:books: Documentation preview :books:: https://meltano-sdk--2058.org.readthedocs.build/en/2058/

<!-- readthedocs-preview meltano-sdk end -->

codspeed-hq[bot]: ## [CodSpeed Performance Report](https://codspeed.io/meltano/sdk/branches/BuzzCutNorman:refactor-add-singerwriter-class-to-io-base)

### Merging #2058 will **not alter performance**

<sub>Comparing <code>BuzzCutNorman:refactor-add-singerwriter-class-to-io-base</code> (ea8f7e8) with <code>main</code> (5b84912)</sub>



### Summary

`✅ 1` untouched benchmarks





codecov[bot]: ## [Codecov](https://app.codecov.io/gh/meltano/sdk/pull/2058?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:
> Comparison is base [(`5b84912`)](https://app.codecov.io/gh/meltano/sdk/commit/5b849128a28689087b5d8bfa8406ce615ed92eb6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) 87.45% compared to head [(`ea8f7e8`)](https://app.codecov.io/gh/meltano/sdk/pull/2058?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) 87.44%.


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


```diff
@@            Coverage Diff             @@
##             main    #2058      +/-   ##
==========================================
- Coverage   87.45%   87.44%   -0.01%     
==========================================
  Files          58       58              
  Lines        4861     4860       -1     
  Branches      763      763              
==========================================
- Hits         4251     4250       -1     
  Misses        429      429              
  Partials      181      181              
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/meltano/sdk/pull/2058?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/io_base.py b/singer_sdk/io_base.py
index d389bfd0..f7c2ed66 100644
--- a/singer_sdk/io_base.py
+++ b/singer_sdk/io_base.py
@@ -10,7 +10,9 @@ import sys
 import typing as t
 from collections import Counter, defaultdict
 
-from singer_sdk._singerlib import SingerMessageType
+from singer_sdk._singerlib.messages import Message, SingerMessageType
+from singer_sdk._singerlib.messages import format_message as singer_format_message
+from singer_sdk._singerlib.messages import write_message as singer_write_message
 from singer_sdk.helpers._compat import final
 
 logger = logging.getLogger(__name__)
@@ -143,3 +145,26 @@ class SingerReader(metaclass=abc.ABCMeta):
 
     def _process_endofpipe(self) -> None:
         logger.debug("End of pipe reached")
+
+
+class SingerWriter:
+    """Interface for all plugins writting Singer messages to stdout."""
+
+    def format_message(self, message: Message) -> str:
+        """Format a message as a JSON string.
+
+        Args:
+            message: The message to format.
+
+        Returns:
+            The formatted message.
+        """
+        return singer_format_message(message)
+
+    def write_message(self, message: Message) -> None:
+        """Write a message to stdout.
+
+        Args:
+            message: The message to write.
+        """
+        singer_write_message(message)
diff --git a/singer_sdk/mapper_base.py b/singer_sdk/mapper_base.py
index b0be198b..2cc943a4 100644
--- a/singer_sdk/mapper_base.py
+++ b/singer_sdk/mapper_base.py
@@ -7,14 +7,16 @@ import typing as t
 
 import click
 
-import singer_sdk._singerlib as singer
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.helpers.capabilities import CapabilitiesEnum, PluginCapabilities
-from singer_sdk.io_base import SingerReader
+from singer_sdk.io_base import SingerReader, SingerWriter
 from singer_sdk.plugin_base import PluginBase
 
+if t.TYPE_CHECKING:
+    import singer_sdk._singerlib as singer
 
-class InlineMapper(PluginBase, SingerReader, metaclass=abc.ABCMeta):
+
+class InlineMapper(PluginBase, SingerReader, SingerWriter, metaclass=abc.ABCMeta):
     """Abstract base class for inline mappers."""
 
     @classproperty
@@ -28,10 +30,9 @@ class InlineMapper(PluginBase, SingerReader, metaclass=abc.ABCMeta):
             PluginCapabilities.STREAM_MAPS,
         ]
 
-    @staticmethod
-    def _write_messages(messages: t.Iterable[singer.Message]) -> None:
+    def _write_messages(self, messages: t.Iterable[singer.Message]) -> None:
         for message in messages:
-            singer.write_message(message)
+            self.write_message(message)
 
     def _process_schema_message(self, message_dict: dict) -> None:
         self._write_messages(self.map_schema_message(message_dict))
diff --git a/singer_sdk/streams/core.py b/singer_sdk/streams/core.py
index 9463b2f3..f306f1ec 100644
--- a/singer_sdk/streams/core.py
+++ b/singer_sdk/streams/core.py
@@ -769,7 +769,7 @@ class Stream(metaclass=abc.ABCMeta):
         if (not self._is_state_flushed) and (
             self.tap_state != self._last_emitted_state
         ):
-            singer.write_message(singer.StateMessage(value=self.tap_state))
+            self._tap.write_message(singer.StateMessage(value=self.tap_state))
             self._last_emitted_state = copy.deepcopy(self.tap_state)
             self._is_state_flushed = True
 
@@ -797,7 +797,7 @@ class Stream(metaclass=abc.ABCMeta):
     def _write_schema_message(self) -> None:
         """Write out a SCHEMA message with the stream schema."""
         for schema_message in self._generate_schema_messages():
-            singer.write_message(schema_message)
+            self._tap.write_message(schema_message)
 
     @property
     def mask(self) -> singer.SelectionMask:
@@ -849,7 +849,7 @@ class Stream(metaclass=abc.ABCMeta):
             record: A single stream record.
         """
         for record_message in self._generate_record_messages(record):
-            singer.write_message(record_message)
+            self._tap.write_message(record_message)
 
         self._is_state_flushed = False
 
@@ -864,7 +864,7 @@ class Stream(metaclass=abc.ABCMeta):
             encoding: The encoding to use for the batch.
             manifest: A list of filenames for the batch.
         """
-        singer.write_message(
+        self._tap.write_message(
             SDKBatchMessage(
                 stream=self.name,
                 encoding=encoding,
diff --git a/singer_sdk/tap_base.py b/singer_sdk/tap_base.py
index f7a31e73..2f0f5874 100644
--- a/singer_sdk/tap_base.py
+++ b/singer_sdk/tap_base.py
@@ -11,7 +11,7 @@ from enum import Enum
 
 import click
 
-from singer_sdk._singerlib import Catalog, StateMessage, write_message
+from singer_sdk._singerlib import Catalog, StateMessage
 from singer_sdk.configuration._dict_config import merge_missing_config_jsonschema
 from singer_sdk.exceptions import AbortedSyncFailedException, AbortedSyncPausedException
 from singer_sdk.helpers import _state
@@ -25,6 +25,7 @@ from singer_sdk.helpers.capabilities import (
     PluginCapabilities,
     TapCapabilities,
 )
+from singer_sdk.io_base import SingerWriter
 from singer_sdk.plugin_base import PluginBase
 
 if t.TYPE_CHECKING:
@@ -45,7 +46,7 @@ class CliTestOptionValue(Enum):
     Disabled = "disabled"
 
 
-class Tap(PluginBase, metaclass=abc.ABCMeta):
+class Tap(PluginBase, SingerWriter, metaclass=abc.ABCMeta):
     """Abstract base class for taps.
 
     The Tap class governs configuration, validation, and stream discovery for tap
@@ -439,7 +440,7 @@ class Tap(PluginBase, metaclass=abc.ABCMeta):
         """Sync all streams."""
         self._reset_state_progress_markers()
         self._set_compatible_replication_methods()
-        write_message(StateMessage(value=self.state))
+        self.write_message(StateMessage(value=self.state))
 
         stream: Stream
         for stream in self.streams.values():


Test patch:
diff --git a/tests/_singerlib/test_messages.py b/tests/_singerlib/test_messages.py
index b4a33db4..e1025949 100644
--- a/tests/_singerlib/test_messages.py
+++ b/tests/_singerlib/test_messages.py
@@ -8,7 +8,7 @@ import pytest
 from pytz import timezone
 
 import singer_sdk._singerlib as singer
-from singer_sdk._singerlib.messages import format_message
+from singer_sdk.io_base import SingerWriter
 
 UTC = datetime.timezone.utc
 
@@ -19,22 +19,24 @@ def test_exclude_null_dict():
 
 
 def test_format_message():
+    singerwriter = SingerWriter()
     message = singer.RecordMessage(
         stream="test",
         record={"id": 1, "name": "test"},
     )
-    assert format_message(message) == (
+    assert singerwriter.format_message(message) == (
         '{"type": "RECORD", "stream": "test", "record": {"id": 1, "name": "test"}}'
     )
 
 
 def test_write_message():
+    singerwriter = SingerWriter()
     message = singer.RecordMessage(
         stream="test",
         record={"id": 1, "name": "test"},
     )
     with redirect_stdout(io.StringIO()) as out:
-        singer.write_message(message)
+        singerwriter.write_message(message)
 
     assert out.getvalue() == (
         '{"type": "RECORD", "stream": "test", "record": {"id": 1, "name": "test"}}\n'