Update caching of version metadata

pyproject.toml

@@ -68,7 +68,7 @@ dependencies = [
     "humanize",
     "tabulate",
     "cachier",
-    "pystow>=0.8.16",
+    "pystow>=0.8.17",
     "bioversions>=0.10.45",
     "bioregistry>=0.13.59",
     "ssslm>=0.2.0",

src/pyobo/api/metadata.py

@@ -2,16 +2,15 @@
 
 import logging
 from functools import lru_cache
-from typing import Any, cast
 
+from pystow.cache import CachedPydantic
 from typing_extensions import Unpack
 
-from .utils import get_version_from_kwargs
 from ..constants import GetOntologyKwargs, check_should_force
 from ..getters import get_ontology
 from ..identifier_utils import wrap_norm_prefix
-from ..utils.cache import cached_json
 from ..utils.path import CacheArtifact, get_cache_path
+from ..utils.ver import VersionMetadata, get_version_from_kwargs
 
 __all__ = [
     "get_metadata",
@@ -22,14 +21,14 @@
 
 @lru_cache
 @wrap_norm_prefix
-def get_metadata(prefix: str, **kwargs: Unpack[GetOntologyKwargs]) -> dict[str, Any]:
+def get_metadata(prefix: str, **kwargs: Unpack[GetOntologyKwargs]) -> VersionMetadata:
     """Get metadata for the ontology."""
     version = get_version_from_kwargs(prefix, kwargs)
     path = get_cache_path(prefix, CacheArtifact.metadata, version=version)
 
-    @cached_json(path=path, force=check_should_force(kwargs))
-    def _get_json() -> dict[str, Any]:
+    @CachedPydantic(path=path, force=check_should_force(kwargs), model_cls=VersionMetadata)
+    def _inner() -> VersionMetadata:
         ontology = get_ontology(prefix, **kwargs)
         return ontology.get_metadata()
 
-    return cast(dict[str, Any], _get_json())
+    return _inner()

src/pyobo/api/utils.py

@@ -1,19 +1,18 @@
 """Utilities for high-level API."""
 
-import json
-import logging
-import os
 import warnings
-from functools import lru_cache
-from typing import Literal, cast, overload
 
-import bioversions
 import curies
 from bioregistry import NormalizedNamableReference as Reference
 from curies import ReferenceTuple
 
-from ..constants import GetOntologyKwargs
-from ..utils.path import prefix_directory_join
+from ..utils.ver import (
+    VersionError,
+    get_version,
+    get_version_from_kwargs,
+    get_version_pins,
+    pin_version,
+)
 
 __all__ = [
     "VersionError",
@@ -23,120 +22,6 @@
     "pin_version",
 ]
 
-logger = logging.getLogger(__name__)
-
-
-class VersionError(ValueError):
-    """A catch-all for version getting failure."""
-
-
-# docstr-coverage:excused `overload`
-@overload
-def get_version(prefix: str, *, strict: Literal[True] = ...) -> str: ...
-
-
-# docstr-coverage:excused `overload`
-@overload
-def get_version(prefix: str, *, strict: Literal[False] = ...) -> str | None: ...
-
-
-@lru_cache(None)
-def get_version(prefix: str, *, strict: bool = False) -> str | None:
-    """Get the version for the resource, if available.
-
-    :param prefix: the resource name
-    :param strict: Should an error be raised if no version is available?
-
-    :returns: The version if available else None
-
-    :raises VersionError: if the version is not available and strict mode is enabled
-    """
-    # Prioritize loaded environment variable PYOBO_VERSION_PINS dictionary
-    if version := get_version_pins().get(prefix):
-        return version
-
-    try:
-        version = bioversions.get_version(prefix)
-    except KeyError:
-        pass  # this prefix isn't available from bioversions
-    except Exception as e:
-        msg = f"[{prefix}] could not get version from bioversions"
-        if strict:
-            raise ValueError(msg) from e
-        logger.warning(msg)
-        raise
-    else:
-        if version:
-            return version
-
-    metadata_json_path = prefix_directory_join(prefix, name="metadata.json", ensure_exists=False)
-    if metadata_json_path.exists():
-        data = json.loads(metadata_json_path.read_text())
-        version = cast(str | None, data["version"])
-        if version:
-            return version
-
-    if strict:
-        raise ValueError
-
-    return None
-
-
-def get_version_from_kwargs(prefix: str, kwargs: GetOntologyKwargs) -> str | None:
-    """Get the version for the resource based on generic keyword arguments."""
-    if version := kwargs.get("version"):
-        return version
-    # it's okay if none gets returned after getting this far, we at least tried
-    return get_version(prefix, strict=False)
-
-
-def pin_version(prefix: str, version: str) -> None:
-    """Pin the version."""
-    get_version_pins()[prefix] = version
-
-
-@lru_cache(1)
-def get_version_pins() -> dict[str, str]:
-    """Retrieve user-defined resource version pins.
-
-    To set your own resource pins, set your machine's environmental variable
-    "PYOBO_VERSION_PINS" to a JSON string containing string resource prefixes as keys
-    and string versions of their respective resource as values. Constraining version
-    pins will make PyOBO rely on cached versions of a resource. A user might want to pin
-    resource versions that are used by PyOBO due to the fact that PyOBO will download
-    the latest version of a resource if it is not pinned. This downloading process can
-    lead to a slow-down in downstream applications that rely on PyOBO.
-    """
-    version_pins_str = os.getenv("PYOBO_VERSION_PINS")
-    if not version_pins_str:
-        return {}
-
-    try:
-        version_pins = cast(dict[str, str], json.loads(version_pins_str))
-    except ValueError as e:
-        logger.error(
-            "The value for the environment variable PYOBO_VERSION_PINS "
-            "must be a valid JSON string: %s",
-            e,
-        )
-        return {}
-
-    for prefix, version in list(version_pins.items()):
-        if not isinstance(prefix, str) or not isinstance(version, str):
-            logger.error(f"The prefix:{prefix} and version:{version} name must both be strings")
-            del version_pins[prefix]
-
-    logger.debug(
-        f"These are the resource versions that are pinned.\n"
-        f"{version_pins}. "
-        f"\nPyobo will download the latest version of a resource if it's "
-        f"not pinned.\nIf you want to use a specific version of a "
-        f"resource, edit your PYOBO_VERSION_PINS environmental "
-        f"variable which is a JSON string to include a prefix and version "
-        f"name."
-    )
-    return version_pins
-
 
 def _get_pi(
     prefix: str | curies.Reference | ReferenceTuple, identifier: str | None = None, /

src/pyobo/cli/database.py

@@ -152,10 +152,16 @@ def metadata(zenodo: bool, directory: Path, **kwargs: Unpack[DatabaseKwargs]) ->
     def _iter_metadata_internal(
         **kwargs: Unpack[IterHelperHelperDict],
     ) -> Iterable[tuple[str, str, str, bool]]:
-        for prefix, data in iter_helper_helper(get_metadata, **kwargs):
-            version = data["version"]
-            logger.debug(f"[{prefix}] using version {version}")
-            yield prefix, version, data["date"], bioregistry.is_deprecated(prefix)
+        for prefix, metadata in iter_helper_helper(get_metadata, **kwargs):
+            if metadata.version is None and metadata.date is None:
+                continue
+            logger.debug(f"[{prefix}] using version {metadata.version}")
+            yield (
+                prefix,
+                metadata.version or "",
+                metadata.date.isoformat() if metadata.date else "",
+                bioregistry.is_deprecated(prefix),
+            )
 
     it = _iter_metadata_internal(**kwargs)
     db_output_helper(

src/pyobo/cli/lookup.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import inspect
-import json
 from collections.abc import Callable, Iterable, Mapping
 from typing import TYPE_CHECKING, ParamSpec, TypeVar
 
@@ -113,7 +112,7 @@ def metadata(prefix: str, **kwargs: Unpack[GetOntologyKwargs]) -> None:
     from ..api import get_metadata
 
     metadata = get_metadata(prefix, **kwargs)
-    click.echo(json.dumps(metadata, indent=2))
+    click.echo(metadata.model_dump_json(indent=2))
 
 
 @lookup_annotate

src/pyobo/getters.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import datetime
-import json
 import logging
 import pathlib
 import subprocess
@@ -22,6 +21,8 @@
 import pystow.utils
 import requests.exceptions
 from bioregistry.schema import AnnotatedURL, RDFFormat
+from pydantic import BaseModel
+from pystow.utils import write_pydantic_json
 from tabulate import tabulate
 from tqdm.auto import tqdm
 from typing_extensions import Unpack
@@ -551,17 +552,13 @@ def db_output_helper(
             detailed_summary_writer.writerows((*keys, v) for keys, v in c_detailed.most_common())
         rv.append(("Summary (Detailed)", db_summary_detailed_path))
 
-    with open(db_metadata_path, "w") as file:
-        json.dump(
-            {
-                "version": get_version(),
-                "git_hash": get_git_hash(),
-                "date": datetime.datetime.now().strftime("%Y-%m-%d-%H-%M"),
-                "count": sum(c.values()),
-            },
-            file,
-            indent=2,
-        )
+    database_metadata = DatabaseMetadata(
+        version=get_version(),
+        git_hash=get_git_hash(),
+        date=datetime.datetime.now(),
+        count=sum(c.values()),
+    )
+    write_pydantic_json(database_metadata, db_metadata_path, indent=2)
 
     elapsed = time.time() - start
     click.secho(f"\nWrote the following files in {elapsed:.1f} seconds\n", fg="green")
@@ -572,3 +569,12 @@ def db_output_helper(
     click.echo()
 
     return [path for _, path in rv]
+
+
+class DatabaseMetadata(BaseModel):
+    """A model for database metadata."""
+
+    version: str  # PyOBO version
+    git_hash: str  # PyOBO git hash
+    date: datetime.datetime
+    count: int

src/pyobo/sources/hgnc/hgncgenefamily.py

@@ -71,7 +71,7 @@ def get_gene_family_terms(*, version: str | None = None, force: bool = False) ->
     yield from terms
 
 
-def _get_terms_helper(version: str, force: bool = False) -> Iterable[Term]:
+def _get_terms_helper(version: str, *, force: bool = False) -> Iterable[Term]:
     alias_df = ensure_df(
         GENE_GROUP_PREFIX, url=FAMILIES_ALIAS_URL, force=force, sep=",", version=version
     )

src/pyobo/sources/mesh.py

@@ -12,11 +12,12 @@
 import bioversions
 from lxml import etree
 from lxml.etree import Element
+from pystow.cache import CachedJSON
 from tqdm.auto import tqdm
 
 from pyobo.identifier_utils import standardize_ec
 from pyobo.struct import Obo, Reference, Synonym, Term, default_reference
-from pyobo.utils.cache import cached_json, cached_mapping
+from pyobo.utils.cache import cached_mapping
 from pyobo.utils.path import ensure_path, prefix_directory_join
 
 __all__ = [
@@ -173,7 +174,7 @@ def ensure_mesh_descriptors(
 ) -> list[Mapping[str, Any]]:
     """Get the parsed MeSH dictionary, and cache it if it wasn't already."""
 
-    @cached_json(path=prefix_directory_join(PREFIX, name="desc.json", version=version), force=force)
+    @CachedJSON(path=prefix_directory_join(PREFIX, name="desc.json", version=version), force=force)
     def _inner() -> list[dict[str, Any]]:
         path = ensure_path(PREFIX, url=get_descriptors_url(version), version=version)
         root = _get_xml_root(path)
@@ -199,7 +200,7 @@ def get_supplemental_url(version: str) -> str:
 def ensure_mesh_supplemental_records(version: str, force: bool = False) -> list[Mapping[str, Any]]:
     """Get the parsed MeSH dictionary, and cache it if it wasn't already."""
 
-    @cached_json(path=prefix_directory_join(PREFIX, name="supp.json", version=version), force=force)
+    @CachedJSON(path=prefix_directory_join(PREFIX, name="supp.json", version=version), force=force)
     def _inner() -> list[dict[str, Any]]:
         path = ensure_path(PREFIX, url=get_supplemental_url(version), version=version)
         root = _get_xml_root(path)

src/pyobo/struct/struct.py

@@ -28,7 +28,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 pystow.utils import safe_open, write_pydantic_json
 from tqdm.auto import tqdm
 
 from . import vocabulary as v
@@ -61,7 +61,6 @@
     _tag_property_targets,
 )
 from .utils import _boolean_tag, obo_escape_slim
-from ..api.utils import get_version
 from ..constants import (
     BUILD_SUBDIRECTORY_NAME,
     DATE_FORMAT,
@@ -81,6 +80,7 @@
     get_relation_cache_path,
     prefix_directory_join,
 )
+from ..utils.ver import VersionMetadata, get_version
 from ..version import get_version as get_pyobo_version
 
 __all__ = [
@@ -1259,7 +1259,7 @@ def _get_cache_path(self, name: CacheArtifact) -> Path:
 
     @property
     def _root_metadata_path(self) -> Path:
-        return prefix_directory_join(self.ontology, name="metadata.json")
+        return prefix_directory_join(self.ontology, name=CacheArtifact.metadata.value)
 
     @property
     def _obo_path(self) -> Path:
@@ -1329,8 +1329,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, operation="write") as file:
-                json.dump(metadata, file, indent=2)
+            write_pydantic_json(metadata, path, indent=2)
 
     def write_prefix_map(self) -> None:
         """Write a prefix map file that includes all prefixes used in this ontology."""
@@ -1588,12 +1587,9 @@ def to_obonet(self: Obo, *, use_tqdm: bool = False) -> nx.MultiDiGraph:
         )
         return rv
 
-    def get_metadata(self) -> dict[str, Any]:
+    def get_metadata(self) -> VersionMetadata:
         """Get metadata."""
-        return {
-            "version": self.data_version,
-            "date": self.date and self.date.isoformat(),
-        }
+        return VersionMetadata(version=self.data_version, date=self.date)
 
     def iterate_references(self, *, use_tqdm: bool = False) -> Iterable[Reference]:
         """Iterate over identifiers."""