Merge pull request #57 from linkml/ruff_lint_autofixes

Automatic python code fixes suggested by ruff

Author: ialarmedalien

src/linkml_map/compiler/compiler.py

@@ -81,9 +81,9 @@ def _compile_header(self, specification: TransformationSpecification) -> str:
     def _compile_iterator(self, specification: TransformationSpecification) -> Iterator[str]:
         raise NotImplementedError
 
-    def derived_target_schemaview(self, specification: TransformationSpecification):
+    def derived_target_schemaview(self, specification: TransformationSpecification) -> SchemaView:
         """
-        Returns a view over the target schema, including any derived classes.
+        Return a view over the target schema, including any derived classes.
         """
         mapper = SchemaMapper(source_schemaview=self.source_schemaview)
         return SchemaView(yaml_dumper.dumps(mapper.derive_schema(specification)))

src/linkml_map/compiler/graphviz_compiler.py

@@ -20,10 +20,10 @@ class Record(BaseModel):
     fields: list[tuple[str, str]] = []
 
     @property
-    def id(self):
+    def id(self) -> str:
         return f"{self.source}{self.name}"
 
-    def __str__(self):
+    def __str__(self) -> str:
         return (
             f"""<
         <TABLE BORDER="0" CELLBORDER="1" CELLSPACING="0">
@@ -47,6 +47,7 @@ class GraphvizObject(CompiledSpecification):
     def render(self, file_path: str, format="png", view=False) -> None:
         """
         Render a graphviz graph to a file.
+
         :param file_path:
         :return:
         """

src/linkml_map/compiler/j2_based_compiler.py

@@ -24,11 +24,13 @@ def compile(self, specification: TransformationSpecification) -> CompiledSpecifi
         if not template_dir:
             template_dir = TEMPLATE_DIR
         if not template_dir:
-            raise ValueError("template_dir must be set")
+            msg = "template_dir must be set"
+            raise ValueError(msg)
         loader = FileSystemLoader(template_dir)
         env = Environment(loader=loader, autoescape=True)
         if not self.template_name:
-            raise ValueError("template_name must be set")
+            msg = "template_name must be set"
+            raise ValueError(msg)
         template = env.get_template(self.template_name)
         rendered = template.render(
             spec=specification,

src/linkml_map/compiler/python_compiler.py

@@ -95,10 +95,7 @@ def _compile_iterator(self, specification: TransformationSpecification) -> Itera
 
     def _compiled_class_derivations_iter(self, cd: ClassDerivation) -> Iterator[str]:
         sv = self.source_schemaview
-        if cd.populated_from:
-            populated_from = cd.populated_from
-        else:
-            populated_from = cd.name
+        populated_from = cd.populated_from if cd.populated_from else cd.name
         if populated_from not in sv.all_classes():
             return
         induced_slots = {s.name: s for s in sv.class_induced_slots(populated_from)}

src/linkml_map/functions/unit_conversion.py

@@ -9,7 +9,7 @@
 
 from enum import Enum
 from functools import lru_cache
-from typing import Any, Optional
+from typing import Any, Optional, Union
 
 import lark
 import pint
@@ -55,7 +55,7 @@ class DimensionalityError(Exception):
 
 def convert_units(
     magnitude: float, from_unit: str, to_unit: str, system: Optional[UnitSystem] = None
-):
+) -> Any:
     """
     Convert a quantity between units.
 
@@ -79,28 +79,32 @@ def convert_units(
     :param to_unit:
     :return: converted magnitude
     """
-    import pint
-
     ureg: pint.UnitRegistry = get_unit_registry(system)
     from_unit = normalize_unit(from_unit, system)
     to_unit = normalize_unit(to_unit, system)
     try:
         from_unit_q = ureg.parse_units(from_unit)
-    except lark.exceptions.UnexpectedCharacters:
-        raise UndefinedUnitError(f"Cannot parse unit: {from_unit}")
-    except pint.errors.UndefinedUnitError:
-        raise UndefinedUnitError(f"Unknown source unit: {from_unit}")
+    except lark.exceptions.UnexpectedCharacters as err:
+        msg = f"Cannot parse unit: {from_unit}"
+        raise UndefinedUnitError(msg) from err
+    except pint.errors.UndefinedUnitError as err:
+        msg = f"Unknown source unit: {from_unit}"
+        raise UndefinedUnitError(msg) from err
     quantity = magnitude * from_unit_q
     try:
         return quantity.to(to_unit).magnitude
-    except pint.errors.UndefinedUnitError:
-        raise UndefinedUnitError(f"Unknown target unit: {from_unit}")
-    except pint.errors.DimensionalityError:
-        raise DimensionalityError(f"Cannot convert from {from_unit} to {to_unit}")
+    except pint.errors.UndefinedUnitError as err:
+        msg = f"Unknown target unit: {from_unit}"
+        raise UndefinedUnitError(msg) from err
+    except pint.errors.DimensionalityError as err:
+        msg = f"Cannot convert from {from_unit} to {to_unit}"
+        raise DimensionalityError(msg) from err
 
 
 @lru_cache
-def get_unit_registry(system: Optional[UnitSystem] = None) -> Any:
+def get_unit_registry(
+    system: Optional[UnitSystem] = None,
+) -> Union[pint.UnitRegistry, PintUcumRegistry]:
     """
     Get a unit registry.
 
@@ -140,18 +144,21 @@ def get_unit_registry(system: Optional[UnitSystem] = None) -> Any:
     if system.value in dir(ureg.sys):
         ureg.default_system = system.value
         return ureg
-    raise NotImplementedError(f"Unknown unit system: {system}")
+    msg = f"Unknown unit system: {system}"
+    raise NotImplementedError(msg)
 
 
 def normalize_unit(unit: str, system: Optional[UnitSystem] = None) -> str:
-    if system is None:
-        return unit
-    if system == UnitSystem.UCUM:
-        try:
-            return str(get_unit_registry(system).from_ucum(unit))
-        except pint.errors.UndefinedUnitError:
-            raise UndefinedUnitError(f"Unknown unit: {unit}")
-        except lark.exceptions.UnexpectedCharacters:
-            raise UndefinedUnitError(f"Cannot parse unit: {unit}")
-    else:
+    """Normalize the unit to UnitSystem.UCUM, if possible."""
+    if system is None or system != UnitSystem.UCUM:
         return unit
+
+    # this is UnitSystem.UCUM
+    try:
+        return str(get_unit_registry(system).from_ucum(unit))
+    except pint.errors.UndefinedUnitError as err:
+        msg = f"Unknown unit: {unit}"
+        raise UndefinedUnitError(msg) from err
+    except lark.exceptions.UnexpectedCharacters as err:
+        msg = f"Cannot parse unit: {unit}"
+        raise UndefinedUnitError(msg) from err

src/linkml_map/importer/importer.py

@@ -27,7 +27,7 @@ class Importer(ABC):
     @abstractmethod
     def import_specification(self, input: Any) -> TransformationSpecification:
         """
-        Import into a specification
+        Import into a specification.
 
         :param specification:
         :return:

src/linkml_map/inference/inference.py

@@ -1,11 +1,13 @@
+"""Infer missing values in a specification."""
+
 from linkml_runtime import SchemaView
 
 from linkml_map.datamodel.transformer_model import TransformationSpecification
 
 
 def induce_missing_values(
     specification: TransformationSpecification, source_schemaview: SchemaView
-):
+) -> None:
     """
     Infer missing values in a specification.
 
@@ -24,15 +26,14 @@ def induce_missing_values(
             # TODO: decide if this is the desired behavior
             if sd.populated_from is None and sd.expr is None:
                 sd.populated_from = sd.name
-            if not sd.range:
+            if not sd.range and sd.populated_from:
                 # auto-populate range field
-                if sd.populated_from:
-                    if cd.populated_from not in source_schemaview.all_classes():
-                        continue
-                    source_induced_slot = source_schemaview.induced_slot(
-                        sd.populated_from, cd.populated_from
-                    )
-                    source_induced_slot_range = source_induced_slot.range
-                    for range_cd in specification.class_derivations.values():
-                        if range_cd.populated_from == source_induced_slot_range:
-                            sd.range = range_cd.name
+                if cd.populated_from not in source_schemaview.all_classes():
+                    continue
+                source_induced_slot = source_schemaview.induced_slot(
+                    sd.populated_from, cd.populated_from
+                )
+                source_induced_slot_range = source_induced_slot.range
+                for range_cd in specification.class_derivations.values():
+                    if range_cd.populated_from == source_induced_slot_range:
+                        sd.range = range_cd.name

src/linkml_map/inference/inverter.py

@@ -1,3 +1,5 @@
+"""Invert a transformation specification."""
+
 import logging
 import re
 from copy import copy
@@ -19,8 +21,8 @@
 logger = logging.getLogger(__name__)
 
 
-class NonInvertibleSpecification(ValueError):
-    pass
+class NonInvertibleSpecificationError(ValueError):
+    """Error thrown when a specification is not invertible."""
 
 
 @dataclass
@@ -40,7 +42,7 @@ class TransformationSpecificationInverter:
 
     strict: bool = field(default=True)
 
-    def invert(self, spec: TransformationSpecification):
+    def invert(self, spec: TransformationSpecification) -> TransformationSpecification:
         """
         Invert a transformation specification.
 
@@ -57,7 +59,9 @@ def invert(self, spec: TransformationSpecification):
             inverted_spec.enum_derivations[inverted_ed.name] = inverted_ed
         return inverted_spec
 
-    def invert_class_derivation(self, cd: ClassDerivation, spec: TransformationSpecification):
+    def invert_class_derivation(
+        self, cd: ClassDerivation, spec: TransformationSpecification
+    ) -> ClassDerivation:
         """
         Invert a class derivation.
 
@@ -73,10 +77,13 @@ def invert_class_derivation(self, cd: ClassDerivation, spec: TransformationSpeci
             if inverted_sd:
                 inverted_cd.slot_derivations[inverted_sd.name] = inverted_sd
             elif self.strict:
-                raise NonInvertibleSpecification(f"Cannot invert slot derivation: {sd.name}")
+                msg = f"Cannot invert slot derivation: {sd.name}"
+                raise NonInvertibleSpecificationError(msg)
         return inverted_cd
 
-    def invert_enum_derivation(self, ed: EnumDerivation, spec: TransformationSpecification):
+    def invert_enum_derivation(
+        self, ed: EnumDerivation, spec: TransformationSpecification
+    ) -> EnumDerivation:
         """
         Invert an enum derivation.
 
@@ -88,7 +95,8 @@ def invert_enum_derivation(self, ed: EnumDerivation, spec: TransformationSpecifi
             name=ed.populated_from if ed.populated_from else ed.name, populated_from=ed.name
         )
         if inverted_ed.expr:
-            raise NonInvertibleSpecification("TODO: invert enum derivation with expression")
+            msg = "TODO: invert enum derivation with expression"
+            raise NonInvertibleSpecificationError(msg)
         for pv_deriv in ed.permissible_value_derivations.values():
             inverted_pv_deriv = PermissibleValueDerivation(
                 name=pv_deriv.populated_from if pv_deriv.populated_from else pv_deriv.name,
@@ -116,15 +124,13 @@ def invert_slot_derivation(
                 if not self.strict:
                     return None
                 # TODO: add logic for reversible expressions
-                raise NonInvertibleSpecification(
-                    f"Cannot invert expression {sd.expr} in slot derivation: {sd.name}"
-                )
+                msg = f"Cannot invert expression {sd.expr} in slot derivation: {sd.name}"
+                raise NonInvertibleSpecificationError(msg)
+
         if not populated_from:
             # use defaults. TODO: decide on semantics of defaults
             populated_from = sd.name
-            # raise NonInvertibleSpecification(f"No populate_from or expr in slot derivation: {sd.name}")
         inverted_sd = SlotDerivation(name=populated_from, populated_from=sd.name)
-        # source_cls_name = spec.class_derivations[cd.populated_from].name
         source_cls_name = cd.populated_from
         if (
             source_cls_name is None or source_cls_name in self.source_schemaview.all_classes()
@@ -133,7 +139,6 @@ def invert_slot_derivation(
         else:
             source_slot = None
         if sd.range:
-            # source_slot = self.source_schemaview.induced_slot(sd.populated_from, source_cls_name)
             inverted_sd.range = source_slot.range
             if source_slot.range in self.source_schemaview.all_classes():
                 id_slot = self.source_schemaview.get_identifier_slot(
@@ -144,14 +149,13 @@ def invert_slot_derivation(
         if source_slot and source_slot.multivalued:
             if source_slot.inlined_as_list:
                 inverted_sd.cast_collection_as = CollectionType.MultiValuedList
-            elif source_slot.inlined:
-                if source_slot.range in self.source_schemaview.all_classes():
-                    id_slot = self.source_schemaview.get_identifier_slot(
-                        source_slot.range, use_key=True
-                    )
-                    if id_slot:
-                        inverted_sd.cast_collection_as = CollectionType.MultiValuedDict
-                        inverted_sd.dictionary_key = id_slot.name
+            elif source_slot.inlined and source_slot.range in self.source_schemaview.all_classes():
+                id_slot = self.source_schemaview.get_identifier_slot(
+                    source_slot.range, use_key=True
+                )
+                if id_slot:
+                    inverted_sd.cast_collection_as = CollectionType.MultiValuedDict
+                    inverted_sd.dictionary_key = id_slot.name
         if sd.unit_conversion:
             source_slot = self.source_schemaview.induced_slot(sd.populated_from, source_cls_name)
             target_unit = None
@@ -165,8 +169,6 @@ def invert_slot_derivation(
             inverted_uc = UnitConversionConfiguration(
                 target_unit=target_unit, target_unit_scheme=target_unit_scheme
             )
-            # if sd.unit_conversion.target_unit:
-            #    inverted_uc.source_unit = sd.unit_conversion.target_unit
             if sd.unit_conversion.source_unit_slot:
                 inverted_uc.target_unit_slot = sd.unit_conversion.source_unit_slot
             if sd.unit_conversion.source_magnitude_slot: