Safe read/write

src/pystow/utils/__init__.py

@@ -101,6 +101,8 @@
     safe_open_dict_reader,
     safe_open_json,
     safe_open_yaml,
+    safe_read_text,
+    safe_write_text,
     write_json,
     write_yaml,
 )
@@ -192,7 +194,9 @@
     "safe_open_reader",
     "safe_open_writer",
     "safe_open_yaml",
+    "safe_read_text",
     "safe_tarfile_open",
+    "safe_write_text",
     "safe_zipfile_open",
     "stream_write_pydantic_jsonl",
     "tarfile_writestr",

src/pystow/utils/safe_open.py

@@ -35,6 +35,8 @@
     "safe_open_dict_reader",
     "safe_open_json",
     "safe_open_yaml",
+    "safe_read_text",
+    "safe_write_text",
     "write_json",
     "write_yaml",
 ]
@@ -192,6 +194,29 @@ def safe_open_yaml(
         return yaml.safe_load(file)
 
 
+def safe_write_text(
+    s: str,
+    path: str | Path | TextIO,
+    *,
+    encoding: str | None = None,
+    newline: str | None = None,
+) -> int:
+    """Write text to a file."""
+    with _open_write_text(path, encoding=encoding, newline=newline) as file:
+        return file.write(s)
+
+
+def safe_read_text(
+    path: str | Path | TextIO,
+    *,
+    encoding: str | None = None,
+    newline: str | None = None,
+) -> str:
+    """Read text from a file."""
+    with _open_read_text(path, encoding=encoding, newline=newline) as file:
+        return file.read()
+
+
 def write_yaml(
     data: Any,
     path: str | Path | TextIO,

tests/test_utils/test_utils.py

@@ -52,6 +52,8 @@
     safe_open_json,
     safe_open_reader,
     safe_open_writer,
+    safe_read_text,
+    safe_write_text,
     tarfile_writestr,
     write_lzma_csv,
     write_tarfile_csv,
@@ -463,6 +465,15 @@ def test_encodings(self) -> None:
                 with safe_open(path, encoding=encoding, operation="read") as file:
                     self.assertEqual(TEST_TXT_CONTENT, file.read(), msg=f"failed for {encoding}")
 
+    def test_safe_write(self) -> None:
+        """Test safe write."""
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory).joinpath("test.txt")
+            safe_write_text(TEST_TXT_CONTENT, path)
+            self.assertEqual(TEST_TXT_CONTENT, safe_read_text(path))
+            with path.open() as file:
+                self.assertEqual(TEST_TXT_CONTENT, file.read())
+
     def test_gzip(self) -> None:
         """Test gzipping."""
         with tempfile.TemporaryDirectory() as tmpdir: