diff --git a/src/linkml_map/inference/schema_mapper.py b/src/linkml_map/inference/schema_mapper.py index e554d97..4398714 100644 --- a/src/linkml_map/inference/schema_mapper.py +++ b/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