Reorganize I/O functions

src/semra/database.py

@@ -1,6 +1,5 @@
 """Assemble a database."""
 
-import csv
 import subprocess
 import time
 from collections.abc import Iterable
@@ -29,6 +28,7 @@
     write_pickle,
     write_sssom,
 )
+from semra.io.io_utils import safe_open_writer
 from semra.pipeline import REFRESH_SOURCE_OPTION, UPLOAD_OPTION
 from semra.sources import SOURCE_RESOLVER
 from semra.sources.wikidata import get_wikidata_mappings_by_prefix
@@ -338,8 +338,7 @@ def _get_pickle_path(subdirectory: str, key: str) -> Path:
 
 
 def _write_summary() -> None:
-    with SUMMARY_PATH.open("w") as file:
-        writer = csv.writer(file, delimiter="\t")
+    with safe_open_writer(SUMMARY_PATH) as writer:
         writer.writerow(("prefix", "mappings", "seconds", "source_type"))
         for prefix, n_mappings, time_delta, source_type in summaries:
             writer.writerow((prefix, n_mappings, round(time_delta, 2), source_type))

src/semra/io/io.py

@@ -2,14 +2,12 @@
 
 from __future__ import annotations
 
-import contextlib
-import csv
 import gzip
 import logging
 import pickle
 import typing as t
 import uuid
-from collections.abc import Generator, Iterable
+from collections.abc import Iterable
 from pathlib import Path
 from typing import Any, TextIO, cast
 
@@ -22,7 +20,7 @@
 from tqdm.autonotebook import tqdm
 from tqdm.contrib.logging import logging_redirect_tqdm
 
-from .io_utils import get_confidence_str, get_name_by_curie
+from .io_utils import get_confidence_str, get_name_by_curie, safe_open_writer
 from ..rules import UNSPECIFIED_MAPPING
 from ..struct import Evidence, Mapping, MappingSet, ReasonedEvidence, Reference, SimpleEvidence
 
@@ -517,28 +515,8 @@ def write_sssom(
     df.to_csv(file, sep="\t", index=False)
 
 
-@contextlib.contextmanager
-def _safe_opener(path: str | Path, read: bool = False) -> Generator[TextIO, None, None]:
-    path = Path(path).expanduser().resolve()
-    if path.suffix.endswith(".gz"):
-        with gzip.open(path, mode="rt" if read else "wt") as file:
-            yield file
-    else:
-        with open(path, mode="r" if read else "w") as file:
-            yield file
-
-
-@contextlib.contextmanager
-def _safe_writer(f: str | Path | TextIO):  # type:ignore
-    if isinstance(f, str | Path):
-        with _safe_opener(f, read=False) as file:
-            yield csv.writer(file, delimiter="\t")
-    else:
-        yield csv.writer(f, delimiter="\t")
-
-
 def _write_sssom_stream(mappings: Iterable[Mapping], file: str | Path | TextIO) -> None:
-    with _safe_writer(file) as writer:
+    with safe_open_writer(file) as writer:
         writer.writerow(SSSOM_DEFAULT_COLUMNS)
         writer.writerows(
             _get_sssom_row(m, e)

src/semra/io/io_utils.py

@@ -2,8 +2,13 @@
 
 from __future__ import annotations
 
+import contextlib
+import csv
+import gzip
+from collections.abc import Generator
 from functools import cache
-from typing import cast
+from pathlib import Path
+from typing import TextIO, cast
 
 import bioregistry
 import pyobo
@@ -15,6 +20,8 @@
     "get_confidence_str",
     "get_name_by_curie",
     "get_orcid_name",
+    "safe_open",
+    "safe_open_writer",
 ]
 
 SKIP_PREFIXES = {
@@ -67,3 +74,25 @@ def get_confidence_str(x: ConfidenceMixin) -> str:
     """Safely get a confidence from an evidence."""
     confidence = x.get_confidence()
     return str(round(confidence, CONFIDENCE_PRECISION))
+
+
+@contextlib.contextmanager
+def safe_open(path: str | Path, read: bool = False) -> Generator[TextIO, None, None]:
+    """Safely open a file for reading or writing text."""
+    path = Path(path).expanduser().resolve()
+    if path.suffix.endswith(".gz"):
+        with gzip.open(path, mode="rt" if read else "wt") as file:
+            yield file
+    else:
+        with open(path, mode="r" if read else "w") as file:
+            yield file
+
+
+@contextlib.contextmanager
+def safe_open_writer(f: str | Path | TextIO, *, delimiter: str = "\t"):  # type:ignore
+    """Open a CSV writer, wrapping :func:`csv.writer`."""
+    if isinstance(f, str | Path):
+        with safe_open(f, read=False) as file:
+            yield csv.writer(file, delimiter=delimiter)
+    else:
+        yield csv.writer(f, delimiter=delimiter)