feat: implement support for copy directives

As of now CopyDirective objects were supported by the specification, but
the implementation to handle them was missing.
This commit adds support for them.

Signed-off-by: Silvano Cirujano Cuesta <[email protected]>

src/linkml_map/inference/schema_mapper.py

@@ -51,6 +51,56 @@ class SchemaMapper:
 
     slot_info: Dict[Tuple[str, str], Any] = field(default_factory=lambda: {})
 
+    def _copy_elements(
+        self,
+        specification: [TransformationSpecification, ClassDerivation],
+        source: [SchemaDefinition, ClassDefinition],
+        copy_directive: CopyDirective,
+        target: [SchemaDefinition, ClassDefinition],
+        element_type: str,
+    ) -> SchemaDefinition:
+        if element_type == "class":
+            plural = "classes"
+        else:
+            plural = element_type + "s"
+        src_elements = getattr(source, plural)
+        tgt_elements = getattr(target, plural)
+        if type(src_elements) is list:
+            for element in src_elements:
+                if (
+                    specification.copy_directives[copy_directive].exclude
+                    and element in specification.copy_directives[copy_directive].exclude
+                ):
+                    continue
+                tgt_elements.append(element)
+        else:
+            for element in src_elements.keys():
+                if (
+                    specification.copy_directives[copy_directive].exclude
+                    and element in specification.copy_directives[copy_directive].exclude
+                ):
+                    continue
+                tgt_elements[element] = src_elements[element]
+        return target
+
+    def _copy_all(
+        self,
+        specification: [TransformationSpecification, ClassDerivation],
+        source: [SchemaDefinition, ClassDefinition],
+        copy_directive: CopyDirective,
+        target: [SchemaDefinition, ClassDefinition],
+        element_types: [str],
+    ) -> SchemaDefinition:
+        for element_type in element_types:
+            target = self._copy_elements(
+                specification,
+                source,
+                copy_directive,
+                target,
+                element_type,
+            )
+        return target
+
     def derive_schema(
         self,
         specification: Optional[TransformationSpecification] = None,
@@ -73,6 +123,27 @@ def derive_schema(
         if target_schema_name is None:
             target_schema_name = source_schema.name + suffix
         target_schema = SchemaDefinition(id=target_schema_id, name=target_schema_name)
+        if hasattr(specification, "copy_directives"):
+            if type(specification.copy_directives) is list:
+                for copy_directive in specification.copy_directives:
+                    if specification.copy_directives[copy_directive].copy_all:
+                        target_schema = self._copy_all(
+                            specification,
+                            source_schema,
+                            copy_directive,
+                            target_schema,
+                            ["class", "slot", "enum"],
+                        )
+            else:
+                for copy_directive in specification.copy_directives.keys():
+                    if specification.copy_directives[copy_directive].copy_all:
+                        target_schema = self._copy_all(
+                            specification,
+                            source_schema,
+                            copy_directive,
+                            target_schema,
+                            ["class", "slot", "enum"],
+                        )
         for im in source_schema.imports:
             target_schema.imports.append(im)
         for prefix in source_schema.prefixes.values():
@@ -112,6 +183,27 @@ def _derive_class(self, class_derivation: ClassDerivation) -> ClassDefinition:
             target_class.slots = []
             target_class.attributes = {}
             target_class.slot_usage = {}
+        if hasattr(class_derivation, "copy_directives"):
+            if type(class_derivation.copy_directives) is list:
+                for copy_directive in class_derivation.copy_directives:
+                    if class_derivation.copy_directives[copy_directive].copy_all:
+                        target_class = self._copy_all(
+                            class_derivation,
+                            source_class,
+                            copy_directive,
+                            target_class,
+                            ["slot", "attribute"],
+                        )
+            else:
+                for copy_directive in class_derivation.copy_directives.keys():
+                    if class_derivation.copy_directives[copy_directive].copy_all:
+                        target_class = self._copy_all(
+                            class_derivation,
+                            source_class,
+                            copy_directive,
+                            target_class,
+                            ["slot", "attribute"],
+                        )
         for slot_derivation in class_derivation.slot_derivations.values():
             slot_definition = self._derive_slot(slot_derivation)
             target_class.attributes[slot_definition.name] = slot_definition