Update

Author: cthoyt

src/pyobo/struct/functional/ontology.py

@@ -8,6 +8,7 @@
 from pathlib import Path
 
 from curies import Converter
+from pystow.utils import safe_open
 from rdflib import OWL, RDF, Graph, term
 
 from pyobo.struct.functional.dsl import Annotation, Annotations, Axiom, Box
@@ -16,7 +17,6 @@
     FunctionalOWLSerializable,
     list_to_funowl,
 )
-from pyobo.utils.io import safe_open
 
 __all__ = [
     "Document",
@@ -108,9 +108,9 @@ def to_rdf(self) -> Graph:
         return graph
 
     def write_funowl(self, path: str | Path) -> None:
-        """Write functional OWL to a file.."""
+        """Write functional OWL to a file."""
         path = Path(path).expanduser().resolve()
-        with safe_open(path, read=False) as file:
+        with safe_open(path, operation="write") as file:
             file.write(self.to_funowl())
 
     def to_funowl(self) -> str:

src/pyobo/struct/obo/reader.py

@@ -18,6 +18,7 @@
 from curies.preprocessing import BlocklistError
 from curies.vocabulary import SynonymScope
 from more_itertools import pairwise
+from pystow.utils import safe_open
 from tqdm.auto import tqdm
 
 from .reader_utils import (
@@ -52,7 +53,6 @@
     get_rules,
 )
 from ...utils.cache import write_gzipped_graph
-from ...utils.io import safe_open
 from ...utils.misc import STATIC_VERSION_REWRITES, cleanup_version
 
 __all__ = [
@@ -90,7 +90,7 @@ def from_obo_path(
                 )
     else:
         logger.info("[%s] parsing OBO with obonet from %s", prefix or "<unknown>", path)
-        with safe_open(path, read=True) as file:
+        with safe_open(path, operation="read") as file:
             graph = _read_obo(file, prefix, ignore_obsolete=ignore_obsolete, use_tqdm=use_tqdm)
 
     if prefix:

src/pyobo/struct/obograph/export.py

@@ -8,11 +8,11 @@
 import obographs as og
 from curies import Converter, ReferenceTuple
 from curies import vocabulary as v
+from pystow.utils import safe_open
 
 from pyobo.identifier_utils.api import get_converter
 from pyobo.struct import Obo, OBOLiteral, Stanza, Term, TypeDef
 from pyobo.struct import typedef as tdv
-from pyobo.utils.io import safe_open
 
 __all__ = [
     "to_obograph",
@@ -25,7 +25,7 @@ def write_obograph(obo: Obo, path: str | Path, *, converter: Converter | None =
     """Write an ontology to a file as OBO Graph JSON."""
     path = Path(path).expanduser().resolve()
     raw_graph = to_obograph(obo, converter=converter)
-    with safe_open(path, read=False) as file:
+    with safe_open(path, operation="write") as file:
         file.write(raw_graph.model_dump_json(indent=2, exclude_none=True, exclude_unset=True))
 
 

src/pyobo/struct/struct.py

@@ -25,6 +25,7 @@
 from curies import Converter, ReferenceTuple
 from curies import vocabulary as _cv
 from more_click import force_option, verbose_option
+from pystow.utils import safe_open
 from tqdm.auto import tqdm
 from typing_extensions import Self
 
@@ -70,7 +71,7 @@
     TARGET_PREFIX,
 )
 from ..utils.cache import write_gzipped_graph
-from ..utils.io import multidict, safe_open, write_iterable_tsv
+from ..utils.io import multidict, write_iterable_tsv
 from ..utils.path import (
     CacheArtifact,
     get_cache_path,
@@ -997,7 +998,7 @@ def write_obo(
                 unit="line",
             )
         if isinstance(file, str | Path | os.PathLike):
-            with safe_open(file, read=False) as fh:
+            with safe_open(file, operation="write") as fh:
                 self._write_lines(it, fh)
         else:
             self._write_lines(it, file)
@@ -1173,7 +1174,7 @@ def write_metadata(self) -> None:
         metadata = self.get_metadata()
         for path in (self._root_metadata_path, self._get_cache_path(CacheArtifact.metadata)):
             logger.debug("[%s] caching metadata to %s", self._prefix_version, path)
-            with safe_open(path, read=False) as file:
+            with safe_open(path, operation="write") as file:
                 json.dump(metadata, file, indent=2)
 
     def write_prefix_map(self) -> None:

src/pyobo/utils/cache.py

@@ -12,8 +12,9 @@
 from pystow.cache import CachedDataFrame as cached_df  # noqa:N813
 from pystow.cache import CachedJSON as cached_json  # noqa:N813
 from pystow.cache import CachedPickle as cached_pickle  # noqa:N813
+from pystow.utils import safe_open
 
-from .io import open_map_tsv, open_multimap_tsv, safe_open, write_map_tsv, write_multimap_tsv
+from .io import open_map_tsv, open_multimap_tsv, write_map_tsv, write_multimap_tsv
 
 __all__ = [
     "cached_collection",
@@ -69,13 +70,13 @@ def dump(self, rv: Mapping[str, str]) -> None:
 
 def get_gzipped_graph(path: str | Path) -> nx.MultiDiGraph:
     """Read a graph that's gzipped nodelink."""
-    with safe_open(path, read=True) as file:
+    with safe_open(path, operation="read") as file:
         return nx.node_link_graph(json.load(file), edges=NODE_LINK_STYLE)
 
 
 def write_gzipped_graph(graph: nx.MultiDiGraph, path: str | Path) -> None:
     """Write a graph as gzipped nodelink."""
-    with safe_open(path, read=False) as file:
+    with safe_open(path, operation="write") as file:
         json.dump(nx.node_link_data(graph, edges=NODE_LINK_STYLE), file)
 
 

src/pyobo/utils/io.py

@@ -1,15 +1,14 @@
 """I/O utilities."""
 
 import collections.abc
-import contextlib
 import csv
 import gzip
 import logging
 from collections import defaultdict
 from collections.abc import Generator, Iterable, Mapping
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Literal, TextIO, TypeVar, cast
+from typing import TypeVar, cast
 
 import pandas as pd
 import pystow.utils
@@ -22,7 +21,6 @@
     "multisetdict",
     "open_map_tsv",
     "open_multimap_tsv",
-    "safe_open",
     "safe_open_writer",
     "write_iterable_tsv",
     "write_map_tsv",
@@ -145,19 +143,3 @@ def write_iterable_tsv(
         if header is not None:
             writer.writerow(header)
         writer.writerows(it)
-
-
-@contextlib.contextmanager
-def safe_open(
-    path: str | Path, read: bool, encoding: str | None = None
-) -> Generator[TextIO, None, None]:
-    """Safely open a file for reading or writing text."""
-    # TODO replace me!
-    path = Path(path).expanduser().resolve()
-    mode: Literal["rt", "wt"] = "rt" if read else "wt"
-    if path.suffix.endswith(".gz"):
-        with gzip.open(path, mode=mode, encoding=encoding) as file:
-            yield file
-    else:
-        with open(path, mode=mode) as file:
-            yield file