Merge branch 'main' into issue-9203_compile_python_module_name

Author: ialarmedalien

linkml_runtime/utils/schemaview.py

@@ -8,6 +8,7 @@
 import uuid
 import warnings
 from collections import defaultdict, deque
+from collections.abc import Iterable
 from copy import copy, deepcopy
 from dataclasses import dataclass
 from enum import Enum
@@ -44,7 +45,7 @@
 from linkml_runtime.utils.pattern import PatternResolver
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Iterable, Mapping
+    from collections.abc import Callable, Mapping
     from types import NotImplementedType
 
     from linkml_runtime.utils.metamodelcore import URI, URIorCURIE
@@ -96,8 +97,13 @@ class OrderedBy(Enum):
 BLACK = 2
 
 
-def detect_cycles(f: Callable[[Any], Iterable[Any] | None], x: Any) -> None:
-    """Detect cycles in a graph, using function `f` to walk the graph, starting at node `x`.
+def detect_cycles(
+    f: Callable[[Any], Iterable[Any] | None],
+    node_list: Iterable[Any],
+) -> None:
+    """Detect cycles in a graph, using function `f` to walk the graph.
+
+    Input is supplied as a list of nodes that are used to populate the `todo` stack.
 
     Uses the classic white/grey/black colour coding algorithm to track which nodes have been explored. In this
     case, "node" refers to any element in a schema and "neighbours" are elements that can be reached from that
@@ -107,21 +113,28 @@ def detect_cycles(f: Callable[[Any], Iterable[Any] | None], x: Any) -> None:
     GREY:  node is being processed; processing includes exploring all neighbours reachable via f(node)
     BLACK: node and all of its neighbours (and their neighbours, etc.) have been processed
 
-    A directed cycle reachable from node `x` raises a ValueError.
+    A directed cycle reachable from a node or its neighbours raises a ValueError.
 
     :param f: function that returns an iterable of neighbouring nodes (parents or children)
     :type f: Callable[[Any], Iterable[Any] | None]
-    :param x: graph node
-    :type x: Any
-    :raises ValueError: if a cycle is discovered through repeated calls to f(x)
+    :param node_list: list or other iterable of values to process
+    :type node_list: Iterable[Any]
+    :raises ValueError: if a cycle is discovered through repeated calls to f(node)
     """
+    # ensure we have some nodes to start the analysis
+    if not node_list or not isinstance(node_list, Iterable) or isinstance(node_list, str):
+        err_msg = "detect_cycles requires a list of values to process"
+        raise ValueError(err_msg)
+
     # keep track of the processing state of nodes in the graph
     processing_state: dict[Any, int] = {}
 
     # Stack entries are (node, processed_flag).
     # processed_flag == True means all neighbours (nodes generated by running `f(node)`)
     # have been added to the todo stack and the node can be marked BLACK.
-    todo: list[tuple[Any, bool]] = [(x, False)]
+
+    # initialise the todo stack with entries set to False
+    todo: list[tuple[Any, bool]] = [(node, False) for node in node_list]
 
     while todo:
         node, processed_flag = todo.pop()
@@ -173,7 +186,7 @@ def _closure(
     :rtype: list[str | ElementName | ClassDefinitionName | EnumDefinitionName | SlotDefinitionName | TypeDefinitionName]
     """
     if kwargs and kwargs.get("detect_cycles"):
-        detect_cycles(f, x)
+        detect_cycles(f, [x])
 
     rv = [x] if reflexive else []
     visited = []
@@ -452,8 +465,7 @@ def imports_closure(
             # visit item
             sn = todo.pop()
             if sn not in self.schema_map:
-                imported_schema = self.load_import(sn)
-                self.schema_map[sn] = imported_schema
+                self.schema_map[sn] = self.load_import(sn)
 
             # resolve item's imports if it has not been visited already
             # we will get duplicates, but not cycles this way, and
@@ -886,7 +898,9 @@ def _parents(self, e: Element, imports: bool = True, mixins: bool = True, is_a:
     def class_parents(
         self, class_name: CLASS_NAME, imports: bool = True, mixins: bool = True, is_a: bool = True
     ) -> list[ClassDefinitionName]:
-        """:param class_name: child class name
+        """Get the parents of a class.
+
+        :param class_name: child class name
         :param imports: include import closure
         :param mixins: include mixins (default is True)
         :return: all direct parent class names (is_a and mixins)
@@ -1139,11 +1153,24 @@ def type_ancestors(
             **kwargs,
         )
 
+    @lru_cache(None)
+    def type_roots(self, imports: bool = True) -> list[TypeDefinitionName]:
+        """Return all types that have no parents.
+
+        :param imports: whether or not to include imports, defaults to True
+        :type imports: bool, optional
+        :return: list of all root types
+        :rtype: list[TypeDefinitionName]
+        """
+        return [t for t in self.all_types(imports=imports) if not self.type_parents(t, imports=imports)]
+
     @lru_cache(None)
     def enum_parents(
         self, enum_name: ENUM_NAME, imports: bool = False, mixins: bool = False, is_a: bool = True
     ) -> list[EnumDefinitionName]:
-        """:param enum_name: child enum name
+        """Get the parents of an enum.
+
+        :param enum_name: child enum name
         :param imports: include import closure (False)
         :param mixins: include mixins (default is False)
         :return: all direct parent enum names (is_a and mixins)
@@ -1180,12 +1207,20 @@ def enum_ancestors(
             **kwargs,
         )
 
-    @lru_cache(None)
+    @deprecated("Use `permissible_value_parents` instead")
     def permissible_value_parent(
         self, permissible_value: str, enum_name: ENUM_NAME
     ) -> list[str | PermissibleValueText]:
-        """:param enum_name: child enum name
+        return self.permissible_value_parents(permissible_value, enum_name)
+
+    @lru_cache(None)
+    def permissible_value_parents(
+        self, permissible_value: str, enum_name: ENUM_NAME
+    ) -> list[str | PermissibleValueText]:
+        """Get the parents of a permissible value.
+
         :param permissible_value: permissible value
+        :param enum_name: enum for which this is a permissible value
         :return: all direct parent enum names (is_a)
         """
         enum = self.get_enum(enum_name, strict=True)
@@ -1199,8 +1234,10 @@ def permissible_value_parent(
     def permissible_value_children(
         self, permissible_value: str, enum_name: ENUM_NAME
     ) -> list[str | PermissibleValueText]:
-        """:param enum_name: parent enum name
+        """Get the children of a permissible value.
+
         :param permissible_value: permissible value
+        :param enum_name: enum for which this is a permissible value
         :return: all direct child permissible values (is_a)
         """
         enum = self.get_enum(enum_name, strict=True)
@@ -1238,7 +1275,7 @@ def permissible_value_ancestors(
         :rtype: list[str]
         """
         return _closure(
-            lambda x: self.permissible_value_parent(x, enum_name),
+            lambda x: self.permissible_value_parents(x, enum_name),
             permissible_value_text,
             reflexive=reflexive,
             depth_first=depth_first,
@@ -1266,17 +1303,6 @@ def permissible_value_descendants(
             **kwargs,
         )
 
-    @lru_cache(None)
-    def type_roots(self, imports: bool = True) -> list[TypeDefinitionName]:
-        """Return all types that have no parents.
-
-        :param imports: whether or not to include imports, defaults to True
-        :type imports: bool, optional
-        :return: list of all root types
-        :rtype: list[TypeDefinitionName]
-        """
-        return [t for t in self.all_types(imports=imports) if not self.type_parents(t, imports=imports)]
-
     @lru_cache(None)
     def is_multivalued(self, slot_name: SlotDefinition) -> bool:
         """Return True if slot is multivalued, else returns False.
@@ -1818,6 +1844,33 @@ def get_type_designator_slot(self, cn: CLASS_NAME, imports: bool = True) -> Slot
                 return s
         return None
 
+    @lru_cache(None)
+    def _get_string_type(self) -> TypeDefinition:
+        """Get the type used for representing strings.
+
+        This can be used for (e.g.) retrieving the appropriate type for a slot where the range is an enum.
+
+        The method assumes that the string type will either be called "string" or
+        will have the URI "xsd:string", as is the case for the "string" type in linkml:types.
+
+        This method throws an error if there is anything other than one type that fits the criteria.
+
+        :return: the "string" type object
+        :rtype: TypeDefinition
+        """
+        str_type = self.get_type("string")
+        if str_type:
+            return str_type
+
+        # if there isn't a type named "string", search for a type with the URI xsd:string
+        str_type_arr = [v for v in self.all_types().values() if v.uri == "xsd:string"]
+        if len(str_type_arr) == 1:
+            return str_type_arr[0]
+
+        # zero or more than one potential "string" type found
+        err_msg = f"Cannot find a suitable 'string' type: no types with name 'string' {'and more than one type with' if str_type_arr else 'or'} uri 'xsd:string'."
+        raise ValueError(err_msg)
+
     def is_inlined(self, slot: SlotDefinition, imports: bool = True) -> bool:
         """Return true if slot is inferred or asserted inline.
 
@@ -1850,6 +1903,10 @@ def slot_applicable_range_elements(self, slot: SlotDefinition) -> list[ClassDefi
         :param slot:
         :return: list of element types
         """
+        if not slot or not isinstance(slot, SlotDefinition):
+            err_msg = "A SlotDefinition must be provided to generate the slot applicable range elements."
+            raise ValueError(err_msg)
+
         is_any = False
         range_types = []
         for r in self.slot_range_as_union(slot):
@@ -1876,6 +1933,10 @@ def slot_range_as_union(self, slot: SlotDefinition) -> list[ElementName]:
         :param slot:
         :return: list of ranges
         """
+        if not slot or not isinstance(slot, SlotDefinition):
+            err_msg = "A SlotDefinition must be provided to generate the slot range as union."
+            raise ValueError(err_msg)
+
         return list({y.range for y in [slot, *[x for x in [*slot.exactly_one_of, *slot.any_of] if x.range]]})
 
     def induced_slot_range(self, slot: SlotDefinition, strict: bool = False) -> set[str | ElementName]:  # noqa: FBT001, FBT002
@@ -1893,6 +1954,9 @@ def induced_slot_range(self, slot: SlotDefinition, strict: bool = False) -> set[
         :return: set of ranges
         :rtype: set[str | ElementName]
         """
+        if not slot or not isinstance(slot, SlotDefinition):
+            err_msg = "A SlotDefinition must be provided to generate the induced slot range."
+            raise ValueError(err_msg)
 
         slot_range = slot.range
         any_of_range = {x.range for x in slot.any_of if x.range}