Merge pull request #269 from linkml/issue-1404-load-as-dict

Add `load_as_dict` method to Loader class

Author: cmungall

linkml_runtime/loaders/delimited_file_loader.py

@@ -19,34 +19,45 @@ class DelimitedFileLoader(Loader, ABC):
     def delimiter(self):
         pass
 
+    def load_as_dict(self, 
+                     source: str,
+                     index_slot: SlotDefinitionName = None,
+                     schema: SchemaDefinition = None,
+                     schemaview: SchemaView = None,
+                     **kwargs) -> Union[dict, List[dict]]:
+        json_str = self._get_json_str_to_load(source, index_slot, schema, schemaview, **kwargs)
+        return JSONLoader().load_as_dict(json_str)
 
     def load_any(self, *args, **kwargs) -> Union[YAMLRoot, List[YAMLRoot]]:
         return self.load(*args, **kwargs)
 
-
     def loads(self, input,
               target_class: Type[Union[BaseModel, YAMLRoot]],
               index_slot: SlotDefinitionName = None,
               schema: SchemaDefinition = None,
               schemaview: SchemaView = None,
               **kwargs) -> str:
-        if schemaview is None:
-            schemaview = SchemaView(schema)
-        configmap = get_configmap(schemaview, index_slot)
-        config = GlobalConfig(key_configs=configmap, csv_delimiter=self.delimiter)
-        objs = unflatten_from_csv(input, config=config, **kwargs)
-        return JSONLoader().loads(json.dumps({index_slot: objs}), target_class=target_class)
+        json_str = self._get_json_str_to_load(input, index_slot, schema, schemaview, **kwargs)
+        return JSONLoader().loads(json_str, target_class=target_class)
 
     def load(self, source: str,
              target_class: Type[Union[BaseModel, YAMLRoot]],
              index_slot: SlotDefinitionName = None,
              schema: SchemaDefinition = None,
              schemaview: SchemaView = None,
              **kwargs) -> str:
+        json_str = self._get_json_str_to_load(source, index_slot, schema, schemaview, **kwargs)
+        return JSONLoader().loads(json_str, target_class=target_class)
+
+    def _get_json_str_to_load(self,
+                          input,
+                          index_slot: SlotDefinitionName = None,
+                          schema: SchemaDefinition = None,
+                          schemaview: SchemaView = None,
+                          **kwargs):
         if schemaview is None:
             schemaview = SchemaView(schema)
         configmap = get_configmap(schemaview, index_slot)
         config = GlobalConfig(key_configs=configmap, csv_delimiter=self.delimiter)
-        print(f'Loading from {source}')
-        objs = unflatten_from_csv(source, config=config, **kwargs)
-        return JSONLoader().loads(json.dumps({index_slot: objs}), target_class=target_class)
+        objs = unflatten_from_csv(input, config=config, **kwargs)
+        return json.dumps({index_slot: objs})

linkml_runtime/loaders/json_loader.py

@@ -10,20 +10,27 @@
 
 class JSONLoader(Loader):
 
-    def load_any(self, source: Union[str, dict, TextIO], target_class: Type[Union[BaseModel, YAMLRoot]], *, base_dir: Optional[str] = None,
-             metadata: Optional[FileInfo] = None, **_) -> Union[BaseModel, YAMLRoot, List[BaseModel], List[YAMLRoot]]:
-        def loader(data: Union[str, dict], _: FileInfo) -> Optional[Dict]:
-            data_as_dict = json.loads(data) if isinstance(data, str) else data
-            if isinstance(data_as_dict, list):
-                return self.json_clean(data_as_dict)
+    def load_as_dict(self, 
+                     source: Union[str, dict, TextIO], 
+                     *, 
+                     base_dir: Optional[str] = None,
+                     metadata: Optional[FileInfo] = None) -> Union[dict, List[dict]]:
+        data = self._read_source(source, base_dir=base_dir, metadata=metadata, accept_header="application/ld+json, application/json, text/json")
+        data_as_dict = json.loads(data) if isinstance(data, str) else data
+        return self.json_clean(data_as_dict)
+
+    def load_any(self, 
+                 source: Union[str, dict, TextIO], 
+                 target_class: Type[Union[BaseModel, YAMLRoot]], 
+                 *, 
+                 base_dir: Optional[str] = None,
+                 metadata: Optional[FileInfo] = None, 
+                 **_) -> Union[BaseModel, YAMLRoot, List[BaseModel], List[YAMLRoot]]:
+        data_as_dict = self.load_as_dict(source, base_dir=base_dir, metadata=metadata)
+
+        if isinstance(data_as_dict, dict):
             typ = data_as_dict.pop('@type', None)
             if typ and typ != target_class.__name__:
                 logging.warning(f"Warning: input type mismatch. Expected: {target_class.__name__}, Actual: {typ}")
-            return self.json_clean(data_as_dict)
 
-        if not metadata:
-            metadata = FileInfo()
-        if base_dir and not metadata.base_path:
-            metadata.base_path = base_dir
-        return self.load_source(source, loader, target_class,
-                                accept_header="application/ld+json, application/json, text/json", metadata=metadata)
+        return self._construct_target_class(data_as_dict, target_class)

linkml_runtime/loaders/loader_root.py

@@ -52,34 +52,10 @@ def load_source(self,
         :return: Instance of the target class if loader worked
         """
 
-        # Makes coding easier down the line if we've got this, even if it is strictly internal
-        if metadata is None:
-            metadata = FileInfo()
-        if not isinstance(source, dict):
-            data = hbread(source, metadata, metadata.base_path, accept_header)
-        else:
-            data = source
+        data = self._read_source(source, metadata=metadata, base_dir=metadata.base_path, accept_header=accept_header)
         data_as_dict = loader(data, metadata)
+        return self._construct_target_class(data_as_dict, target_class=target_class)
 
-        if data_as_dict:
-            if isinstance(data_as_dict, list):
-               if issubclass(target_class, YAMLRoot):
-                   return [target_class(**as_dict(x)) for x in data_as_dict]
-               elif issubclass(target_class, BaseModel):
-                   return [target_class.parse_obj(**as_dict(x)) for x in data_as_dict]
-               else:
-                   raise ValueError(f'Cannot load list of {target_class}')
-            elif isinstance(data_as_dict, dict):
-                if issubclass(target_class, BaseModel):
-                    return target_class.parse_obj(data_as_dict)
-                else:
-                    return target_class(**data_as_dict)
-            elif isinstance(data_as_dict, JsonObj):
-                return [target_class(**as_dict(x)) for x in data_as_dict]
-            else:
-                raise ValueError(f'Unexpected type {data_as_dict}')
-        else:
-            return None
 
     def load(self, *args, **kwargs) -> Union[BaseModel, YAMLRoot]:
         """
@@ -97,6 +73,9 @@ def load(self, *args, **kwargs) -> Union[BaseModel, YAMLRoot]:
             return results
         else:
             raise ValueError(f'Result is not an instance of BaseModel or YAMLRoot: {type(results)}')
+    
+    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,
@@ -134,3 +113,44 @@ def loads(self, source: str, target_class: Type[Union[BaseModel, YAMLRoot]], *,
         :return: instance of taarget_class
         """
         return self.load(source, target_class, metadata=metadata)
+
+    def _construct_target_class(self, 
+                                data_as_dict: Union[dict, List[dict]],
+                                target_class: Union[Type[YAMLRoot], Type[BaseModel]]) -> Optional[Union[BaseModel, YAMLRoot, List[BaseModel], List[YAMLRoot]]]:
+        if data_as_dict:
+            if isinstance(data_as_dict, list):
+               if issubclass(target_class, YAMLRoot):
+                   return [target_class(**as_dict(x)) for x in data_as_dict]
+               elif issubclass(target_class, BaseModel):
+                   return [target_class.parse_obj(**as_dict(x)) for x in data_as_dict]
+               else:
+                   raise ValueError(f'Cannot load list of {target_class}')
+            elif isinstance(data_as_dict, dict):
+                if issubclass(target_class, BaseModel):
+                    return target_class.parse_obj(data_as_dict)
+                else:
+                    return target_class(**data_as_dict)
+            elif isinstance(data_as_dict, JsonObj):
+                return [target_class(**as_dict(x)) for x in data_as_dict]
+            else:
+                raise ValueError(f'Unexpected type {data_as_dict}')
+        else:
+            return None
+
+    def _read_source(self,
+                     source: Union[str, dict, TextIO], 
+                     *, 
+                     base_dir: Optional[str] = None, 
+                     metadata: Optional[FileInfo] = None, 
+                     accept_header: Optional[str] = "text/plain, application/yaml;q=0.9") -> Union[dict, str]:
+        if metadata is None:
+            metadata = FileInfo()
+        if base_dir and not metadata.base_path:
+            metadata.base_path = base_dir
+
+        if not isinstance(source, dict):
+            data = hbread(source, metadata, metadata.base_path, accept_header)
+        else:
+            data = source
+
+        return data

linkml_runtime/loaders/yaml_loader.py

@@ -14,26 +14,31 @@ class YAMLLoader(Loader):
     A Loader that is capable of instantiating LinkML data objects from a YAML file
     """
 
+    def load_as_dict(self, 
+                     source: Union[str, dict, TextIO], 
+                     *, 
+                     base_dir: Optional[str] = None,
+                     metadata: Optional[FileInfo] = None) -> Union[dict, List[dict]]:
+        if metadata is None:
+            metadata = FileInfo()
+        if base_dir and not metadata.base_path:
+            metadata.base_path = base_dir
+        data = self._read_source(source, base_dir=base_dir, metadata=metadata, accept_header="text/yaml, application/yaml;q=0.9")
+        if isinstance(data, str):
+            data = StringIO(data)
+            if metadata and metadata.source_file:
+                data.name = os.path.relpath(metadata.source_file, metadata.base_path)
+            return yaml.load(data, DupCheckYamlLoader)
+        else:
+            return data
+
     def load_any(self,
                  source: Union[str, dict, TextIO],
                  target_class: Union[Type[YAMLRoot],Type[BaseModel]],
                  *, base_dir: Optional[str] = None,
                  metadata: Optional[FileInfo] = None, **_) -> Union[YAMLRoot, List[YAMLRoot]]:
-        def loader(data: Union[str, dict], source_file: FileInfo) -> Optional[Dict]:
-            if isinstance(data, str):
-                data = StringIO(data)
-                if source_file and source_file.source_file:
-                    data.name = os.path.relpath(source_file.source_file, source_file.base_path)
-                return yaml.load(data, DupCheckYamlLoader)
-            else:
-                return data
-
-        if not metadata:
-            metadata = FileInfo()
-        if base_dir and not metadata.base_path:
-            metadata.base_path = base_dir
-        return self.load_source(source, loader, target_class, accept_header="text/yaml, application/yaml;q=0.9",
-                                metadata=metadata)
+        data_as_dict = self.load_as_dict(source, base_dir=base_dir, metadata=metadata)
+        return self._construct_target_class(data_as_dict, target_class)
 
     def loads_any(self, source: str, target_class: Type[Union[BaseModel, YAMLRoot]], *, metadata: Optional[FileInfo] = None, **_) -> Union[BaseModel, YAMLRoot, List[BaseModel], List[YAMLRoot]]:
         """

tests/test_loaders_dumpers/test_csv_tsv_loader_dumper.py

@@ -62,13 +62,27 @@ def test_csvgen_roundtrip(self):
         logging.debug(f'COMPARE 2: {data}')
         assert roundtrip == data
 
+    def test_csvgen_roundtrip_to_dict(self):
+        schemaview = SchemaView(SCHEMA)
+        data = yaml_loader.load(DATA, target_class=Shop)
+        csv_dumper.dump(data, to_file=OUTPUT, index_slot='all_book_series', schemaview=schemaview)
+        roundtrip = csv_loader.load_as_dict(OUTPUT, index_slot='all_book_series', schemaview=schemaview)
+        assert roundtrip == json_dumper.to_dict(data)
+
     def test_tsvgen_roundtrip(self):
         schemaview = SchemaView(SCHEMA)
         data = yaml_loader.load(DATA, target_class=Shop)
         tsv_dumper.dump(data, to_file=OUTPUT, index_slot='all_book_series', schemaview=schemaview)
         roundtrip = tsv_loader.load(OUTPUT, target_class=Shop, index_slot='all_book_series', schemaview=schemaview)
         assert roundtrip == data
 
+    def test_tsvgen_roundtrip_to_dict(self):
+        schemaview = SchemaView(SCHEMA)
+        data = yaml_loader.load(DATA, target_class=Shop)
+        tsv_dumper.dump(data, to_file=OUTPUT, index_slot='all_book_series', schemaview=schemaview)
+        roundtrip = tsv_loader.load_as_dict(OUTPUT, index_slot='all_book_series', schemaview=schemaview)
+        assert roundtrip == json_dumper.to_dict(data)
+
     def test_csvgen_unroundtrippable(self):
         schemaview = SchemaView(SCHEMA)
         #schema = YAMLGenerator(SCHEMA).schema

tests/test_loaders_dumpers/test_loaders.py

@@ -29,6 +29,16 @@ 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)
 
+    def test_json_load_to_dict(self):
+        data = json_loader.load_as_dict('obo_sample.json', base_dir=self.env.indir)
+        assert isinstance(data, dict)
+        assert "system" in data
+
+    def test_yaml_load_to_dict(self):
+        data = yaml_loader.load_as_dict('obo_sample.yaml', base_dir=self.env.indir)
+        assert isinstance(data, dict)
+        assert "system" in data
+
     @unittest.skipIf(True, "This test will not work until https://github.com/digitalbazaar/pyld/issues/149 is fixed")
     def test_rdf_loader(self):
         """ Load obo_sample.ttl, emit obo_sample_ttl.yaml and check the results