Merge pull request #36 from linkml/just_id_mapping

function to return schema elements by id_prefixes

linkml_runtime/utils/schemaview.py

@@ -6,11 +6,15 @@
 from collections import defaultdict
 from typing import Mapping, Tuple, Type
 from linkml_runtime.utils.namespaces import Namespaces
+from linkml_runtime.utils.context_utils import parse_import_map
+from linkml_runtime.linkml_model.meta import *
 from deprecated.classic import deprecated
-
 from linkml_runtime.utils.context_utils import parse_import_map
 from linkml_runtime.linkml_model.meta import *
 
+logger = logging.getLogger(__name__)
+
+
 MAPPING_TYPE = str  ## e.g. broad, exact, related, ...
 CACHE_SIZE = 1024
 
@@ -622,7 +626,6 @@ def get_element(self, element: Union[ElementName, Element], imports=True) -> Ele
             e = self.get_subset(element, imports=imports)
         return e
 
-
     def get_uri(self, element: Union[ElementName, Element], imports=True, expand=False, native=False) -> str:
         """
         Return the CURIE or URI for a schema element. If the schema defines a specific URI, this is
@@ -677,6 +680,40 @@ def expand_curie(self, uri: str) -> str:
                     return ns[pfx] + local_id
         return uri
 
+    @lru_cache(CACHE_SIZE)
+    def get_elements_applicable_by_identifier(self, identifier: str) -> List[str]:
+        """
+        Get a model element by identifier.  The model element corresponding to the given identifier as available via
+        the id_prefixes mapped to that element.
+
+        :param identifier:
+        :return: Optional[str]
+
+        """
+        elements = self.get_elements_applicable_by_prefix(self.namespaces().prefix_for(identifier))
+        if len(elements) == 0:
+            logger.warning("no element found for the given curie using id_prefixes attribute"
+                           ": %s, try get_mappings method?", identifier)
+        return elements
+
+    @lru_cache(CACHE_SIZE)
+    def get_elements_applicable_by_prefix(self, prefix: str) -> List[str]:
+        """
+        Get a model element by prefix. The model element corresponding to the given prefix as available via
+        the id_prefixes mapped to that element.
+
+        :param prefix: the prefix of a CURIE
+        :return: Optional[str]
+
+        """
+        applicable_elements = []
+        elements = self.all_element()
+        for category, category_element in elements.items():
+            if hasattr(category_element, 'id_prefixes') and prefix in category_element.id_prefixes:
+                applicable_elements.append(category_element.name)
+
+        return applicable_elements
+
     @lru_cache()
     def get_mappings(self, element_name: ElementName = None, imports=True, expand=False) -> Dict[MAPPING_TYPE, List[URIorCURIE]]:
         """

tests/test_utils/input/kitchen_sink_noimports.yaml

@@ -17,6 +17,7 @@ description: |-
 
 default_curi_maps:
   - semweb_context
+  - monarch_context
 prefixes:
   pav: http://purl.org/pav/
   dce: http://purl.org/dc/elements/1.1/
@@ -74,6 +75,11 @@ classes:
       - subset A
     mixins:
       - HasAliases
+    id_prefixes:
+      - ORCID
+      - DOI
+      - ZFIN
+      - WB
     slots:
       - has employment history
       - has familial relationships
@@ -95,6 +101,10 @@ classes:
     is_a: Thing
     mixins:
       - HasAliases
+    id_prefixes:
+      - PMID
+      - ZFIN
+      - WB
 
   Place:
     mixins:

tests/test_utils/test_schemaview.py

@@ -6,7 +6,6 @@
 from linkml_runtime.loaders.yaml_loader import YAMLLoader
 from linkml_runtime.utils.schemaview import SchemaView
 from linkml_runtime.utils.schemaops import roll_up, roll_down
-
 from tests.test_utils import INPUT_DIR
 
 SCHEMA_NO_IMPORTS = os.path.join(INPUT_DIR, 'kitchen_sink_noimports.yaml')
@@ -25,6 +24,21 @@ def test_schemaview(self):
         all_cls = view.all_classes()
         logging.debug(f'n_cls = {len(all_cls)}')
 
+        assert list(view.annotation_dict('is current').values()) == ['bar']
+        logging.debug(view.annotation_dict('employed at'))
+        e = view.get_element('employed at')
+        logging.debug(e.annotations)
+        e = view.get_element('has employment history')
+        logging.debug(e.annotations)
+
+        elements = view.get_elements_applicable_by_identifier("ORCID:1234")
+        assert "Person" in elements
+        elements = view.get_elements_applicable_by_identifier("PMID:1234")
+        assert "Organization" in elements
+        elements = view.get_elements_applicable_by_identifier("http://www.ncbi.nlm.nih.gov/pubmed/1234")
+        assert "Organization" in elements
+        elements = view.get_elements_applicable_by_identifier("TEST:1234")
+        assert "anatomical entity" not in elements
         assert list(view.annotation_dict(SlotDefinitionName('is current')).values()) == ['bar']
         logging.debug(view.annotation_dict(SlotDefinitionName('employed at')))
         element = view.get_element(SlotDefinitionName('employed at'))
@@ -94,6 +108,7 @@ def test_schemaview(self):
         assert view.get_class('agent').class_uri == 'prov:Agent'
         assert view.get_uri('agent') == 'prov:Agent'
         logging.debug(view.get_class('Company').class_uri)
+
         assert view.get_uri('Company') == 'ks:Company'
 
         # test induced slots
@@ -131,7 +146,7 @@ def test_schemaview(self):
 
         a = view.get_class('activity')
         self.assertCountEqual(a.exact_mappings, ['prov:Activity'])
-        logging.debug(view.get_mappings('activity',expand=True))
+        logging.debug(view.get_mappings('activity', expand=True))
         self.assertCountEqual(view.get_mappings('activity')['exact'], ['prov:Activity'])
         self.assertCountEqual(view.get_mappings('activity', expand=True)['exact'], ['http://www.w3.org/ns/prov#Activity'])
 
@@ -168,6 +183,7 @@ def test_rollup_rolldown(self):
         # check to make sure rolled-up classes are deleted
         assert view.class_descendants(element_name, reflexive=False) == []
         roll_down(view, view.class_leaves())
+
         for element_name in view.all_classes():
             c = view.get_class(element_name)
             logging.debug(f'{element_name}')
@@ -177,7 +193,7 @@ def test_rollup_rolldown(self):
         assert 'Thing' not in view.all_classes()
         assert 'Person' not in view.all_classes()
         assert 'Adult' in view.all_classes()
-
+        
     def test_caching(self):
         """
         Determine if cache is reset after modifications made to schema
@@ -239,6 +255,7 @@ def test_imports(self):
         assert view.get_class('agent').class_uri == 'prov:Agent'
         assert view.get_uri('agent') == 'prov:Agent'
         logging.debug(view.get_class('Company').class_uri)
+
         assert view.get_uri('Company') == 'ks:Company'
         assert view.get_uri('Company', expand=True) == 'https://w3id.org/linkml/tests/kitchen_sink/Company'
         logging.debug(view.get_uri("TestClass"))