refactor: Centralize JSON SerDe into helper functions

.flake8

@@ -1,7 +1,7 @@
 [flake8]
 max-line-length = 88
 exclude = cookiecutter
-ignore = E, W
+ignore = E, F, W
 per-file-ignores =
     # Don't require docstrings conventions in private modules
     singer_sdk/helpers/_*.py:DAR

singer_sdk/_singerlib/__init__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from singer_sdk._singerlib import exceptions
 from singer_sdk._singerlib.catalog import (
     Catalog,
     CatalogEntry,
@@ -16,6 +17,7 @@
     SingerMessageType,
     StateMessage,
     exclude_null_dict,
+    format_message,
     write_message,
 )
 from singer_sdk._singerlib.schema import Schema, resolve_schema_references
@@ -35,7 +37,9 @@
     "SingerMessageType",
     "StateMessage",
     "StreamMetadata",
+    "exceptions",
     "exclude_null_dict",
+    "format_message",
     "resolve_schema_references",
     "strftime",
     "strptime_to_utc",

singer_sdk/_singerlib/encoding/__init__.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from .base import GenericSingerReader, GenericSingerWriter, SingerMessageType
+from .simple import SingerReader, SingerWriter
+
+__all__ = [
+    "GenericSingerReader",
+    "GenericSingerWriter",
+    "SingerMessageType",
+    "SingerReader",
+    "SingerWriter",
+]

singer_sdk/_singerlib/encoding/base.py

@@ -0,0 +1,144 @@
+"""Abstract base classes for all Singer messages IO operations."""
+
+from __future__ import annotations
+
+import abc
+import logging
+import typing as t
+from collections import Counter, defaultdict
+
+from singer_sdk._singerlib import exceptions
+from singer_sdk._singerlib.messages import Message, SingerMessageType
+
+logger = logging.getLogger(__name__)
+
+
+# TODO: Use to default to 'str' here
+# https://peps.python.org/pep-0696/
+T = t.TypeVar("T", str, bytes)
+
+
+class GenericSingerReader(t.Generic[T], metaclass=abc.ABCMeta):
+    """Interface for all plugins reading Singer messages as strings or bytes."""
+
+    @t.final
+    def listen(self, file_input: t.IO[T] | None = None) -> None:
+        """Read from input until all messages are processed.
+
+        Args:
+            file_input: Readable stream of messages. Defaults to standard in.
+        """
+        self._process_lines(file_input or self.default_input)
+        self._process_endofpipe()
+
+    def _process_lines(self, file_input: t.IO[T]) -> t.Counter[str]:
+        """Internal method to process jsonl lines from a Singer tap.
+
+        Args:
+            file_input: Readable stream of messages, each on a separate line.
+
+        Returns:
+            A counter object for the processed lines.
+        """
+        stats: dict[str, int] = defaultdict(int)
+        for line in file_input:
+            line_dict = self.deserialize_json(line)
+            self._assert_line_requires(line_dict, requires={"type"})
+
+            record_type: SingerMessageType = line_dict["type"]
+            if record_type == SingerMessageType.SCHEMA:
+                self._process_schema_message(line_dict)
+
+            elif record_type == SingerMessageType.RECORD:
+                self._process_record_message(line_dict)
+
+            elif record_type == SingerMessageType.ACTIVATE_VERSION:
+                self._process_activate_version_message(line_dict)
+
+            elif record_type == SingerMessageType.STATE:
+                self._process_state_message(line_dict)
+
+            elif record_type == SingerMessageType.BATCH:  # pragma: no cover
+                self._process_batch_message(line_dict)
+
+            else:
+                self._process_unknown_message(line_dict)  # pragma: no cover
+
+            stats[record_type] += 1
+
+        return Counter(**stats)
+
+    @property
+    @abc.abstractmethod
+    def default_input(self) -> t.IO[T]: ...
+
+    @staticmethod
+    def _assert_line_requires(line_dict: dict, requires: set[str]) -> None:
+        """Check if dictionary .
+
+        Args:
+            line_dict: TODO
+            requires: TODO
+
+        Raises:
+            InvalidInputLine: raised if any required keys are missing
+        """
+        if not requires.issubset(line_dict):
+            missing = requires - set(line_dict)
+            msg = f"Line is missing required {', '.join(missing)} key(s): {line_dict}"
+            raise exceptions.InvalidInputLine(msg)
+
+    @abc.abstractmethod
+    def deserialize_json(self, line: T) -> dict: ...
+
+    @abc.abstractmethod
+    def _process_schema_message(self, message_dict: dict) -> None: ...
+
+    @abc.abstractmethod
+    def _process_record_message(self, message_dict: dict) -> None: ...
+
+    @abc.abstractmethod
+    def _process_state_message(self, message_dict: dict) -> None: ...
+
+    @abc.abstractmethod
+    def _process_activate_version_message(self, message_dict: dict) -> None: ...
+
+    @abc.abstractmethod
+    def _process_batch_message(self, message_dict: dict) -> None: ...
+
+    def _process_unknown_message(self, message_dict: dict) -> None:  # noqa: PLR6301 # pragma: no cover
+        """Internal method to process unknown message types from a Singer tap.
+
+        Args:
+            message_dict: Dictionary representation of the Singer message.
+
+        Raises:
+            ValueError: raised if a message type is not recognized
+        """
+        record_type = message_dict["type"]
+        msg = f"Unknown message type '{record_type}' in message."
+        raise ValueError(msg)
+
+    def _process_endofpipe(self) -> None:  # noqa: PLR6301
+        logger.debug("End of pipe reached")
+
+
+class GenericSingerWriter(t.Generic[T], metaclass=abc.ABCMeta):
+    """Interface for all plugins writing Singer messages as strings or bytes."""
+
+    def format_message(self, message: Message) -> T:
+        """Format a message as a JSON string.
+
+        Args:
+            message: The message to format.
+
+        Returns:
+            The formatted message.
+        """
+        return self.serialize_json(message.to_dict())
+
+    @abc.abstractmethod
+    def serialize_json(self, obj: object) -> T: ...
+
+    @abc.abstractmethod
+    def write_message(self, message: Message) -> None: ...

singer_sdk/_singerlib/encoding/simple.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import json
+import logging
+import sys
+import typing as t
+
+from singer_sdk._singerlib.exceptions import InvalidInputLine
+from singer_sdk._singerlib.json import deserialize_json, serialize_json
+
+from .base import GenericSingerReader, GenericSingerWriter
+
+if t.TYPE_CHECKING:
+    from singer_sdk._singerlib.messages import Message
+
+logger = logging.getLogger(__name__)
+
+
+class SingerReader(GenericSingerReader[str]):
+    """Base class for all plugins reading Singer messages as strings from stdin."""
+
+    default_input = sys.stdin
+
+    def deserialize_json(self, line: str) -> dict:  # noqa: PLR6301
+        """Deserialize a line of json.
+
+        Args:
+            line: A single line of json.
+
+        Returns:
+            A dictionary of the deserialized json.
+
+        Raises:
+            InvalidInputLine: If the line is not valid JSON.
+        """
+        try:
+            return deserialize_json(line)
+        except json.decoder.JSONDecodeError as exc:
+            logger.exception("Unable to parse:\n%s", line)
+            msg = f"Unable to parse line as JSON: {line}"
+            raise InvalidInputLine(msg) from exc
+
+
+class SingerWriter(GenericSingerWriter[str]):
+    """Interface for all plugins writing Singer messages to stdout."""
+
+    def serialize_json(self, obj: object) -> str:  # noqa: PLR6301
+        """Serialize a dictionary into a line of json.
+
+        Args:
+            obj: A Python object usually a dict.
+
+        Returns:
+            A string of serialized json.
+        """
+        return serialize_json(obj)
+
+    def write_message(self, message: Message) -> None:
+        """Write a message to stdout.
+
+        Args:
+            message: The message to write.
+        """
+        sys.stdout.write(self.format_message(message) + "\n")
+        sys.stdout.flush()

singer_sdk/_singerlib/exceptions.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+__all__ = [
+    "InvalidInputLine",
+]
+
+
+class InvalidInputLine(Exception):
+    """Raised when an input line is not a valid Singer message."""

singer_sdk/_singerlib/json.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import datetime
+import decimal
+import json
+import typing as t
+
+import simplejson
+
+__all__ = [
+    "deserialize_json",
+    "serialize_json",
+]
+
+
+def _default_encoding(obj: t.Any) -> str:  # noqa: ANN401
+    """Default JSON encoder.
+
+    Args:
+        obj: The object to encode.
+
+    Returns:
+        The encoded object.
+    """
+    return obj.isoformat(sep="T") if isinstance(obj, datetime.datetime) else str(obj)
+
+
+def deserialize_json(json_str: str | bytes, **kwargs: t.Any) -> dict:
+    """Deserialize a line of json.
+
+    Args:
+        json_str: A single line of json.
+        **kwargs: Optional key word arguments.
+
+    Returns:
+        A dictionary of the deserialized json.
+    """
+    return json.loads(  # type: ignore[no-any-return]
+        json_str,
+        parse_float=decimal.Decimal,
+        **kwargs,
+    )
+
+
+def serialize_json(obj: object, **kwargs: t.Any) -> str:
+    """Serialize a dictionary into a line of json.
+
+    Args:
+        obj: A Python object usually a dict.
+        **kwargs: Optional key word arguments.
+
+    Returns:
+        A string of serialized json.
+    """
+    return simplejson.dumps(
+        obj,
+        use_decimal=True,
+        default=_default_encoding,
+        separators=(",", ":"),
+        **kwargs,
+    )

singer_sdk/_singerlib/messages.py

@@ -8,7 +8,7 @@
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 
-import simplejson as json
+from singer_sdk._singerlib.json import serialize_json
 
 if sys.version_info < (3, 11):
     from backports.datetime_fromisoformat import MonkeyPatch
@@ -26,18 +26,6 @@ class SingerMessageType(str, enum.Enum):
     BATCH = "BATCH"
 
 
-def _default_encoding(obj: t.Any) -> str:  # noqa: ANN401
-    """Default JSON encoder.
-
-    Args:
-        obj: The object to encode.
-
-    Returns:
-        The encoded object.
-    """
-    return obj.isoformat(sep="T") if isinstance(obj, datetime) else str(obj)
-
-
 def exclude_null_dict(pairs: list[tuple[str, t.Any]]) -> dict[str, t.Any]:
     """Exclude null values from a dictionary.
 
@@ -226,12 +214,7 @@ def format_message(message: Message) -> str:
     Returns:
         The formatted message.
     """
-    return json.dumps(
-        message.to_dict(),
-        use_decimal=True,
-        default=_default_encoding,
-        separators=(",", ":"),
-    )
+    return serialize_json(message.to_dict())
 
 
 def write_message(message: Message) -> None: