Fix version pins

src/pyobo/api/utils.py

@@ -3,18 +3,23 @@
 """Utilities for high-level API."""
 
 import json
+import logging
+import os
+from functools import lru_cache
 from typing import Optional
 
 import bioversions
 
-from ..constants import VERSION_PINS
 from ..utils.path import prefix_directory_join
 
 __all__ = [
     "get_version",
+    "get_version_pins",
     "VersionError",
 ]
 
+logger = logging.getLogger(__name__)
+
 
 class VersionError(ValueError):
     """A catch-all for version getting failure."""
@@ -26,8 +31,8 @@ def get_version(prefix: str) -> Optional[str]:
     :param prefix: the resource name
     :return: The version if available else None
     """
-    # Prioritize loaded environmental variable VERSION_PINS dictionary
-    version = VERSION_PINS.get(prefix)
+    # Prioritize loaded environment variable PYOBO_VERSION_PINS dictionary
+    version = get_version_pins().get(prefix)
     if version:
         return version
     try:
@@ -46,3 +51,46 @@ def get_version(prefix: str) -> Optional[str]:
         return data["version"]
 
     return None
+
+
+@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 = 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

src/pyobo/constants.py

@@ -2,9 +2,7 @@
 
 """Constants for PyOBO."""
 
-import json
 import logging
-import os
 import re
 
 import pystow
@@ -13,7 +11,6 @@
     "RAW_DIRECTORY",
     "DATABASE_DIRECTORY",
     "SPECIES_REMAPPING",
-    "VERSION_PINS",
 ]
 
 logger = logging.getLogger(__name__)
@@ -101,31 +98,3 @@
     "isbn",
     "issn",
 }
-
-# Load version pin dictionary from the environmental variable VERSION_PINS
-try:
-    VERSION_PINS_STR = os.getenv("VERSION_PINS")
-    if not VERSION_PINS_STR:
-        VERSION_PINS = {}
-    else:
-        VERSION_PINS = json.loads(VERSION_PINS_STR)
-        for k, v in VERSION_PINS.items():
-            if not isinstance(k, str) or not isinstance(v, str):
-                logger.error("The prefix and version name must both be " "strings")
-            VERSION_PINS = {}
-            break
-except ValueError as e:
-    logger.error(
-        "The value for the environment variable VERSION_PINS must be a valid JSON string: %s" % e
-    )
-    VERSION_PINS = {}
-
-if VERSION_PINS:
-    logger.debug(
-        f"These are the resource versions that are pinned.\n{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 VERSION_PINS environmental "
-        f"variable which is a JSON string to include a prefix and version "
-        f"name."
-    )

tests/test_version_pins.py

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+
+"""Tests for PyOBO version pins."""
+import os
+import unittest
+from unittest import mock
+
+from pyobo.api.utils import get_version, get_version_pins
+
+MOCK_PYOBO_VERSION_PINS = '{"ncbitaxon": "2024-07-03", "vo":"2024-04-09", "chebi":"235", "bfo":5}'
+FAULTY_MOCK_PYOBO_VERSION_PINS = "{'ncbitaxon': '2024-07-03'}"
+
+
+@mock.patch.dict(os.environ, {"PYOBO_VERSION_PINS": MOCK_PYOBO_VERSION_PINS})
+class TestVersionPins(unittest.TestCase):
+    """Test using user-defined version pins."""
+
+    def setUp(self):
+        """Clear the cache before each test case."""
+        get_version_pins.cache_clear()
+
+    def test_correct_version_pin_types(self):
+        """Test resource and version type."""
+        version_pins = get_version_pins()
+        for resource_prefix, version in version_pins.items():
+            self.assertIsInstance(resource_prefix, str)
+            self.assertIsInstance(version, str)
+
+    def test_use_correct_version_pin(self):
+        """Tests correct resource version is used."""
+        version_pins = get_version_pins()
+        for resource_prefix, version in version_pins.items():
+            self.assertEqual(get_version(resource_prefix), version)
+
+    @mock.patch.dict(os.environ, {"PYOBO_VERSION_PINS": ""})
+    def test_empty_version_pins(self):
+        """Test empty version pins are processed correctly."""
+        version_pins = get_version_pins()
+        self.assertFalse(version_pins)
+
+    @mock.patch.dict(os.environ, {"PYOBO_VERSION_PINS": FAULTY_MOCK_PYOBO_VERSION_PINS})
+    def test_incorrectly_set_version_pins(self):
+        """Test erroneously set version pins are processed correctly."""
+        version_pins = get_version_pins()
+        self.assertFalse(version_pins)