Merge pull request #159 from linkml/doc_ordering

add method to return all_classes, all_slots in alphabetical and/or rank order.

linkml_runtime/utils/schemaview.py

@@ -1,6 +1,7 @@
 import os
 import uuid
 import logging
+import collections
 from functools import lru_cache
 from copy import copy, deepcopy
 from collections import defaultdict, OrderedDict
@@ -9,14 +10,13 @@
 from deprecated.classic import deprecated
 from linkml_runtime.utils.context_utils import parse_import_map
 from linkml_runtime.linkml_model.meta import *
-
+from enum import Enum
 logger = logging.getLogger(__name__)
 
 
 MAPPING_TYPE = str  ## e.g. broad, exact, related, ...
 CACHE_SIZE = 1024
 
-
 SLOTS = 'slots'
 CLASSES = 'classes'
 ENUMS = 'enums'
@@ -30,6 +30,12 @@
 ENUM_NAME = Union[EnumDefinitionName, str]
 
 
+class OrderedBy(Enum):
+    RANK = "rank"
+    LEXICAL = "lexical"
+    PRESERVE = "preserve"
+
+
 def _closure(f, x, reflexive=True, depth_first=True, **kwargs):
     if reflexive:
         rv = [x]
@@ -51,7 +57,6 @@ def _closure(f, x, reflexive=True, depth_first=True, **kwargs):
                 if v not in rv:
                     rv.append(v)
     return rv
-    #return list(OrderedDict.fromkeys(rv))
 
 
 def load_schema_wrap(path: str, **kwargs):
@@ -199,13 +204,60 @@ def all_class(self, imports=True) -> Dict[ClassDefinitionName, ClassDefinition]:
         """
         return self._get_dict(CLASSES, imports)
 
+    def _order_lexically(self, elements: dict):
+        """
+        :param element: slots or class type to order
+        :param imports
+        :return: all classes or slots sorted lexically in schema view
+        """
+        ordered_list_of_names = []
+        ordered_elements = {}
+        for c in elements:
+            ordered_list_of_names.append(c)
+        ordered_list_of_names.sort()
+        for name in ordered_list_of_names:
+            ordered_elements[self.get_element(name).name] = self.get_element(name)
+        return ordered_elements
+
+    def _order_rank(self, elements: dict):
+        """
+        :param elements: slots or classes to order
+        :return: all classes or slots sorted by their rank in schema view
+        """
+
+        rank_map = {}
+        unranked_map = {}
+        rank_ordered_elements = {}
+        for name, definition in elements.items():
+            if definition.rank is None:
+                unranked_map[self.get_element(name).name] = self.get_element(name)
+
+            else:
+                rank_map[definition.rank] = name
+        rank_ordered_map = collections.OrderedDict(sorted(rank_map.items()))
+        for k, v in rank_ordered_map.items():
+            rank_ordered_elements[self.get_element(v).name] = self.get_element(v)
+
+        rank_ordered_elements.update(unranked_map)
+        return rank_ordered_elements
+
     @lru_cache()
-    def all_classes(self, imports=True) -> Dict[ClassDefinitionName, ClassDefinition]:
+    def all_classes(self, ordered_by=OrderedBy.PRESERVE, imports=True) -> Dict[ClassDefinitionName, ClassDefinition]:
         """
+        :param ordered_by: an enumerated parameter that returns all the slots in the order specified.
         :param imports: include imports closure
         :return: all classes in schema view
         """
-        return self._get_dict(CLASSES, imports)
+        classes = copy(self._get_dict(CLASSES, imports))
+
+        if ordered_by == OrderedBy.LEXICAL:
+            ordered_classes = self._order_lexically(elements=classes)
+        elif ordered_by == OrderedBy.RANK:
+            ordered_classes = self._order_rank(elements=classes)
+        else:  # else preserve the order in the yaml
+            ordered_classes = classes
+
+        return ordered_classes
 
     @deprecated("Use `all_slots` instead")
     @lru_cache()
@@ -217,19 +269,29 @@ def all_slot(self, **kwargs) -> Dict[SlotDefinitionName, SlotDefinition]:
         return self.all_slots(**kwargs)
 
     @lru_cache()
-    def all_slots(self, imports=True, attributes=True) -> Dict[SlotDefinitionName, SlotDefinition]:
+    def all_slots(self, ordered_by=OrderedBy.PRESERVE, imports=True, attributes=True) -> Dict[SlotDefinitionName, SlotDefinition]:
         """
+        :param ordered_by: an enumerated parameter that returns all the slots in the order specified.
         :param imports: include imports closure
+        :param attributes: include attributes as slots or not, default is to include.
         :return: all slots in schema view
         """
+
         slots = copy(self._get_dict(SLOTS, imports))
         if attributes:
             for c in self.all_classes().values():
                 for aname, a in c.attributes.items():
                     if aname not in slots:
                         slots[aname] = a
-        return slots
 
+        if ordered_by == OrderedBy.LEXICAL:
+            ordered_slots = self._order_lexically(elements=slots)
+        elif ordered_by == OrderedBy.RANK:
+            ordered_slots = self._order_rank(elements=slots)
+        else:
+            # preserve order in YAML
+            ordered_slots = slots
+        return ordered_slots
 
     @deprecated("Use `all_enums` instead")
     @lru_cache()
@@ -289,11 +351,11 @@ def all_element(self, imports=True) -> Dict[ElementName, Element]:
         :param imports: include imports closure
         :return: all elements in schema view
         """
-        all_classes = self.all_classes(imports)
-        all_slots = self.all_slots(imports)
-        all_enums = self.all_enums(imports)
-        all_types = self.all_types(imports)
-        all_subsets = self.all_subsets(imports)
+        all_classes = self.all_classes(imports=imports)
+        all_slots = self.all_slots(imports=imports)
+        all_enums = self.all_enums(imports=imports)
+        all_types = self.all_types(imports=imports)
+        all_subsets = self.all_subsets(imports=imports)
         # {**a,**b} syntax merges dictionary a and b into a single dictionary, removing duplicates.
         return {**all_classes, **all_slots, **all_enums, **all_types, **all_subsets}
 
@@ -303,11 +365,11 @@ def all_elements(self, imports=True) -> Dict[ElementName, Element]:
         :param imports: include imports closure
         :return: all elements in schema view
         """
-        all_classes = self.all_classes(imports)
-        all_slots = self.all_slots(imports)
-        all_enums = self.all_enums(imports)
-        all_types = self.all_types(imports)
-        all_subsets = self.all_subsets(imports)
+        all_classes = self.all_classes(imports=imports)
+        all_slots = self.all_slots(imports=imports)
+        all_enums = self.all_enums(imports=imports)
+        all_types = self.all_types(imports=imports)
+        all_subsets = self.all_subsets(imports=imports)
         # {**a,**b} syntax merges dictionary a and b into a single dictionary, removing duplicates.
         return {**all_classes, **all_slots, **all_enums, **all_types, **all_subsets}
 
@@ -318,6 +380,7 @@ def _get_dict(self, slot_name: str, imports=True) -> Dict:
         for s in schemas:
             # get the value of element name from the schema, if empty, return empty dictionary.
             d1 = getattr(s, slot_name, {})
+            # {**d,**d1} syntax merges dictionary a and b into a single dictionary, removing duplicates.
             d = {**d, **d1}
 
         return d
@@ -382,7 +445,7 @@ def get_class(self, class_name: CLASS_NAME, imports=True, strict=False) -> Class
         :param imports: include import closure
         :return: class definition
         """
-        c = self.all_classes(imports).get(class_name, None)
+        c = self.all_classes(imports=imports).get(class_name, None)
         if strict and c is None:
             raise ValueError(f'No such class as "{class_name}"')
         else:
@@ -395,9 +458,9 @@ def get_slot(self, slot_name: SLOT_NAME, imports=True, attributes=False, strict=
         :param imports: include import closure
         :return: slot definition
         """
-        slot = self.all_slots(imports).get(slot_name, None)
+        slot = self.all_slots(imports=imports).get(slot_name, None)
         if slot is None and attributes:
-            for c in self.all_classes(imports).values():
+            for c in self.all_classes(imports=imports).values():
                 if slot_name in c.attributes:
                     if slot is not None:
                         # slot name is ambiguous, no results
@@ -488,7 +551,7 @@ def class_children(self, class_name: CLASS_NAME, imports=True, mixins=True, is_a
         :param is_a: include is_a parents (default is True)
         :return: all direct child class names (is_a and mixins)
         """
-        elts = [self.get_class(x) for x in self.all_classes(imports)]
+        elts = [self.get_class(x) for x in self.all_classes(imports=imports)]
         return [x.name for x in elts if (x.is_a == class_name and is_a) or (mixins and class_name in x.mixins)]
 
     @lru_cache()

tests/test_utils/input/kitchen_sink_noimports.yaml

@@ -170,6 +170,7 @@ classes:
     attributes:
       ceo:
         range: Person
+    rank: 3
 
   Dataset:
     attributes:
@@ -203,6 +204,7 @@ classes:
       - prov:Activity
     narrow_mappings:
       - GO:0005198
+    rank: 2
 
   agent:
     description: "a provence-generating agent"
@@ -211,6 +213,7 @@ classes:
         - acted on behalf of
         - was informed by
     class_uri: prov:Agent
+    rank: 1
 
 slots:
   employed at:
@@ -284,15 +287,19 @@ slots:
 
   id:
     identifier: true
+    rank: 1
 
   name:
     required: false
+    rank: 2
 
   description:
+    rank: 3
 
   started at time:
     slot_uri: prov:startedAtTime
     range: date
+    rank: 1
 
   ended at time:
     slot_uri: prov:endedAtTime

tests/test_utils/test_schemaview.py

@@ -7,7 +7,7 @@
 from linkml_runtime.linkml_model.meta import SchemaDefinition, ClassDefinition, SlotDefinitionName, SlotDefinition
 from linkml_runtime.loaders.yaml_loader import YAMLLoader
 from linkml_runtime.utils.introspection import package_schemaview, object_class_definition
-from linkml_runtime.utils.schemaview import SchemaView, SchemaUsage
+from linkml_runtime.utils.schemaview import SchemaView, SchemaUsage, OrderedBy
 from linkml_runtime.utils.schemaops import roll_up, roll_down
 from tests.test_utils import INPUT_DIR
 
@@ -182,6 +182,67 @@ def test_schemaview(self):
             s = view.induced_slot(sn, 'Dataset')
             logging.debug(s)
 
+    def test_all_classes_ordered_lexical(self):
+        view = SchemaView(SCHEMA_NO_IMPORTS)
+        classes = view.all_classes(ordered_by=OrderedBy.LEXICAL)
+
+        ordered_c = []
+        for c in classes.values():
+            ordered_c.append(c.name)
+        assert ordered_c == sorted(ordered_c)
+
+    def test_all_classes_ordered_rank(self):
+        view = SchemaView(SCHEMA_NO_IMPORTS)
+        classes = view.all_classes(ordered_by=OrderedBy.RANK)
+        ordered_c = []
+        for c in classes.values():
+            ordered_c.append(c.name)
+        first_in_line = []
+        second_in_line = []
+        for name, definition in classes.items():
+            if definition.rank == 1:
+                first_in_line.append(name)
+            elif definition.rank == 2:
+                second_in_line.append(name)
+        assert ordered_c[0] in first_in_line
+        assert ordered_c[10] not in second_in_line
+
+    def test_all_classes_ordered_no_ordered_by(self):
+        view = SchemaView(SCHEMA_NO_IMPORTS)
+        classes = view.all_classes()
+        ordered_c = []
+        for c in classes.values():
+            ordered_c.append(c.name)
+        assert "HasAliases" == ordered_c[0]
+        assert "agent" == ordered_c[-1]
+
+    def test_all_slots_ordered_lexical(self):
+        view = SchemaView(SCHEMA_NO_IMPORTS)
+        slots = view.all_slots(ordered_by=OrderedBy.LEXICAL)
+        ordered_s = []
+        for s in slots.values():
+            ordered_s.append(s.name)
+        print(ordered_s)
+        assert ordered_s == sorted(ordered_s)
+
+    def test_all_slots_ordered_rank(self):
+        view = SchemaView(SCHEMA_NO_IMPORTS)
+        slots = view.all_slots(ordered_by=OrderedBy.RANK)
+        ordered_s = []
+        for s in slots.values():
+            ordered_s.append(s.name)
+        print(ordered_s)
+        first_in_line = []
+        second_in_line = []
+        for name, definition in slots.items():
+            if definition.rank == 1:
+                first_in_line.append(name)
+            elif definition.rank == 2:
+                second_in_line.append(name)
+        assert ordered_s[0] in first_in_line
+        assert ordered_s[10] not in second_in_line
+
+
     def test_rollup_rolldown(self):
         # no import schema
         view = SchemaView(SCHEMA_NO_IMPORTS)