Implement naive central JSON loading and dumping functions

singer_sdk/helpers/_util.py

@@ -3,10 +3,53 @@
 from __future__ import annotations
 
 import datetime
-import json
+import decimal
 import typing as t
 from pathlib import Path, PurePath
 
+import simplejson
+
+
+def dump_json(obj: t.Any, **kwargs: t.Any) -> str:  # noqa: ANN401
+    """Dump json data to a file.
+
+    Args:
+        obj: A Python object, usually a dict.
+        **kwargs: Optional key word arguments.
+
+    Returns:
+        A string of serialized json.
+
+    .. warning:: Do not use this function to serialize Singer messages or bulk data.
+                 Use the functions in ``singer_sdk._singerlib.serde`` instead.
+    """
+    return simplejson.dumps(
+        obj,
+        use_decimal=True,
+        separators=(",", ":"),
+        **kwargs,
+    )
+
+
+def load_json(json_str: str, **kwargs: t.Any) -> dict:
+    """Load json data from a file.
+
+    Args:
+        json_str: A valid JSON string.
+        **kwargs: Optional key word arguments.
+
+    Returns:
+        A Python object, usually a dict.
+
+    .. warning:: Do not use this function to parse Singer messages or bulk data.
+                 Use the functions in ``singer_sdk._singerlib.serde`` instead.
+    """
+    return simplejson.loads(  # type: ignore[no-any-return]
+        json_str,
+        parse_float=decimal.Decimal,
+        **kwargs,
+    )
+
 
 def read_json_file(path: PurePath | str) -> dict[str, t.Any]:
     """Read json file, throwing an error if missing."""
@@ -21,7 +64,7 @@ def read_json_file(path: PurePath | str) -> dict[str, t.Any]:
                 msg += f"\nFor more info, please see the sample template at: {template}"
         raise FileExistsError(msg)
 
-    return t.cast(dict, json.loads(Path(path).read_text(encoding="utf-8")))
+    return load_json(Path(path).read_text(encoding="utf-8"))
 
 
 def utc_now() -> datetime.datetime:

singer_sdk/tap_base.py

@@ -10,7 +10,6 @@
 import click
 
 from singer_sdk._singerlib import Catalog, StateMessage
-from singer_sdk._singerlib.serde import serialize_json
 from singer_sdk.configuration._dict_config import merge_missing_config_jsonschema
 from singer_sdk.exceptions import (
     AbortedSyncFailedException,
@@ -20,7 +19,7 @@
 from singer_sdk.helpers import _state
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.helpers._state import write_stream_state
-from singer_sdk.helpers._util import read_json_file
+from singer_sdk.helpers._util import dump_json, read_json_file
 from singer_sdk.helpers.capabilities import (
     BATCH_CONFIG,
     CapabilitiesEnum,
@@ -312,7 +311,7 @@ def catalog_json_text(self) -> str:
         Returns:
             The tap's catalog as formatted JSON text.
         """
-        return serialize_json(self.catalog_dict, indent=2)
+        return dump_json(self.catalog_dict, indent=2)
 
     @property
     def _singer_catalog(self) -> Catalog: