Merge pull request #354 from linkml/issue-2458

add ability to call json_loader with a Path object

Author: cmungall

linkml_runtime/loaders/json_loader.py

@@ -1,6 +1,7 @@
 import json
 import logging
-from typing import Union, TextIO, Optional, Dict, Type, List
+from pathlib import Path
+from typing import Union, TextIO, Optional, Type, List
 
 from hbreader import FileInfo
 
@@ -14,7 +15,7 @@
 class JSONLoader(Loader):
 
     def load_as_dict(self, 
-                     source: Union[str, dict, TextIO], 
+                     source: Union[str, dict, TextIO],
                      *, 
                      base_dir: Optional[str] = None,
                      metadata: Optional[FileInfo] = None) -> Union[dict, List[dict]]:
@@ -23,12 +24,33 @@ def load_as_dict(self,
         return self.json_clean(data_as_dict)
 
     def load_any(self, 
-                 source: Union[str, dict, TextIO], 
+                 source: Union[str, dict, TextIO, Path],
                  target_class: Type[Union[BaseModel, YAMLRoot]], 
                  *, 
                  base_dir: Optional[str] = None,
                  metadata: Optional[FileInfo] = None, 
                  **_) -> Union[BaseModel, YAMLRoot, List[BaseModel], List[YAMLRoot]]:
+        """
+        Load the JSON in source into the python target_class structure
+
+        :param source: JSON data source. Can be a URL, a file name, a JSON string, a resolveable filepath or an existing graph
+        :param target_class: LinkML class to load the JSON into
+        :param base_dir: Base directory that can be used if file name or URL.  This is copied into metadata if present
+        :param metadata: source information. Used by some loaders to record where information came from
+        :return: data instances of target_class
+        """
+
+        # see https://github.com/linkml/linkml/issues/2458, this works around a limitation in hbreader
+        if isinstance(source, Path):
+            full_path = source.resolve()
+            try:
+                with full_path.open("r", encoding="utf-8") as file:
+                    new_source = json.load(file)
+                    source = new_source
+            except FileNotFoundError as e:
+                raise ValueError(f"File not found: {source}") from e
+            except json.JSONDecodeError as e:
+                raise ValueError(f"Invalid JSON in {source}: {e}")
         data_as_dict = self.load_as_dict(source, base_dir=base_dir, metadata=metadata)
 
         if isinstance(data_as_dict, dict):

linkml_runtime/loaders/loader_root.py

@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+from pathlib import Path
 from typing import TextIO, Union, Optional, Callable, Dict, Type, Any, List
 from logging import getLogger
 
@@ -83,7 +84,7 @@ def load_as_dict(self, *args, **kwargs) -> Union[dict, List[dict]]:
         raise NotImplementedError()
 
     @abstractmethod
-    def load_any(self, source: Union[str, dict, TextIO], target_class: Type[Union[BaseModel, YAMLRoot]], *, base_dir: Optional[str] = None,
+    def load_any(self, source: Union[str, dict, TextIO, Path], target_class: Type[Union[BaseModel, YAMLRoot]], *, base_dir: Optional[str] = None,
              metadata: Optional[FileInfo] = None, **_) -> Union[BaseModel, YAMLRoot, List[BaseModel], List[YAMLRoot]]:
         """
         Load source as an instance of target_class, or list of instances of target_class

linkml_runtime/loaders/rdf_loader.py

@@ -1,7 +1,5 @@
 from typing import Union, TextIO, Optional, Type, List
-
 from hbreader import FileInfo
-
 from linkml_runtime.loaders.loader_root import Loader
 from linkml_runtime.utils.context_utils import CONTEXTS_PARAM_TYPE
 from linkml_runtime.utils.yamlutils import YAMLRoot

tests/test_loaders_dumpers/test_loaders.py

@@ -1,7 +1,7 @@
 import os
 import unittest
 from typing import Union, TextIO, Type, Optional
-
+from pathlib import Path
 from hbreader import FileInfo
 
 from linkml_runtime.loaders import yaml_loader, json_loader, rdf_loader, RDFLoader
@@ -25,6 +25,14 @@ def test_yaml_loader(self):
         """ Load obo_sample.yaml, emit obo_sample_yaml.yaml and compare to obo_sample_output.yaml """
         self.loader_test('obo_sample.yaml', Package, yaml_loader)
 
+    def test_json_loader_path(self):
+        """ Load obo_sample.json, emit obo_sample_json.yaml and check the results """
+        REPO_ROOT = Path(__file__).parent.parent.parent
+        path = REPO_ROOT / "tests" /  "test_loaders_dumpers" / "input" / "obo_sample.json"
+        data = json_loader.load(Path(path), Package, base_dir=self.env.indir)
+        assert isinstance(data, Package)
+        assert "system" in data
+
     def test_json_loader(self):
         """ Load obo_sample.json, emit obo_sample_json.yaml and check the results """
         self.loader_test('obo_sample.json', Package, json_loader)