From 5bec3288439006e8dee266a6bf234e5509d52fdb Mon Sep 17 00:00:00 2001 From: Harshad Date: Tue, 23 Jul 2024 15:24:37 -0500 Subject: [PATCH] Added change handler for `kgcl.RemoveNodeFromSubset` (#783) * Added change handler for `kgcl.RemoveNodeFromSubset` * Added type definitions * formatted --- .../utilities/writers/change_handler.py | 81 +++++++++++-------- 1 file changed, 46 insertions(+), 35 deletions(-) diff --git a/src/oaklib/utilities/writers/change_handler.py b/src/oaklib/utilities/writers/change_handler.py index 4597e4fa8..dbad7160e 100644 --- a/src/oaklib/utilities/writers/change_handler.py +++ b/src/oaklib/utilities/writers/change_handler.py @@ -28,7 +28,7 @@ def handle_generic_change( rows = list({row_format.format(change=change) for change in value}) self.write_markdown_table(f"{title}: {len(rows)}", header, rows) - def handle_new_synonym(self, value): + def handle_new_synonym(self, value: List[Change]): # Create rows for the table rows = list( { @@ -43,7 +43,7 @@ def handle_new_synonym(self, value): # Write the "New Synonyms Added" section as a collapsible markdown table self.write_markdown_table(f"Synonyms added: {len(rows)}", header, rows) - def handle_edge_deletion(self, value): + def handle_edge_deletion(self, value: List[Change]): # Create rows for the table rows = list( { @@ -59,7 +59,7 @@ def handle_edge_deletion(self, value): # Write the "Edges Deleted" section as a collapsible markdown table self.write_markdown_table(f"Relationships removed: {len(rows)}", header, rows) - def handle_edge_creation(self, value): + def handle_edge_creation(self, value: List[Change]): # Create rows for the table rows = list( { @@ -75,7 +75,7 @@ def handle_edge_creation(self, value): # Write the "Edges Created" section as a collapsible markdown table self.write_markdown_table(f"Relationships added: {len(rows)}", header, rows) - def handle_edge_change(self, value): + def handle_edge_change(self, value: List[Change]): # Create rows for the table rows = list( { @@ -92,7 +92,7 @@ def handle_edge_change(self, value): # Write the "Edges Changed" section as a collapsible markdown table self.write_markdown_table(f"Relationships changed: {len(rows)}", header, rows) - def handle_node_move(self, value): + def handle_node_move(self, value: List[Change]): # Create rows for the table rows = list( { @@ -109,7 +109,7 @@ def handle_node_move(self, value): # Write the "Nodes Moved" section as a collapsible markdown table self.write_markdown_table(f"Relationships added: {len(rows)}", header, rows) - def handle_predicate_change(self, value): + def handle_predicate_change(self, value: List[Change]): # Create rows for the table rows = list( { @@ -127,7 +127,7 @@ def handle_predicate_change(self, value): # Write the "Predicate Changed" section as a collapsible markdown table self.write_markdown_table(f"Predicates changed: {len(rows)}", header, rows) - def handle_node_rename(self, value): + def handle_node_rename(self, value: List[Change]): # Create rows for the table rows = list( { @@ -142,7 +142,7 @@ def handle_node_rename(self, value): # Write the "Node Renamed" section as a collapsible markdown table self.write_markdown_table(f"Nodes renamed: {len(rows)}", header, rows) - def handle_remove_synonym(self, value): + def handle_remove_synonym(self, value: List[Change]): # Create rows for the table rows = list( { @@ -157,7 +157,7 @@ def handle_remove_synonym(self, value): # Write the "Synonyms Removed" section as a collapsible markdown table self.write_markdown_table(f"Synonyms removed: {len(rows)}", header, rows) - def hand_synonym_predicate_change(self, value): + def hand_synonym_predicate_change(self, value: List[Change]): # Create rows for the table rows = list( { @@ -173,7 +173,7 @@ def hand_synonym_predicate_change(self, value): # Write the "Synonym Predicate Changed" section as a markdown table self.write_markdown_table(f"Synonym predicates changed: {len(rows)}", header, rows) - def handle_node_text_definition_change(self, value): + def handle_node_text_definition_change(self, value: List[Change]): # Create rows for the table rows = list( { @@ -189,7 +189,7 @@ def handle_node_text_definition_change(self, value): # Write the "Node Text Definition Changed" section as a markdown table self.write_markdown_table(f"Text definitions changed: {len(rows)}", header, rows) - def handle_node_text_definition(self, value): + def handle_node_text_definition(self, value: List[Change]): # Create rows for the table rows = list( { @@ -205,7 +205,7 @@ def handle_node_text_definition(self, value): # Write the "Node Text Definition Added" section as a markdown table self.write_markdown_table(f"Text definitions added: {len(rows)}", header, rows) - def handle_node_unobsoletion(self, value): + def handle_node_unobsoletion(self, value: List[Change]): # Create rows for the table rows = list({f"| {self._format_entity_labels(change.about_node)} |" for change in value}) @@ -215,7 +215,7 @@ def handle_node_unobsoletion(self, value): # Write the "Node Unobsoleted" section as a markdown table self.write_markdown_table(f"Nodes unobsoleted: {len(rows)}", header, rows) - def handle_node_creation(self, value): + def handle_node_creation(self, value: List[Change]): # Create rows for the table rows = list({f"| {self._format_entity_labels(change.about_node)} |" for change in value}) @@ -225,7 +225,7 @@ def handle_node_creation(self, value): # Write the "Node Created" section as a markdown table self.write_markdown_table(f"Other nodes added: {len(rows)}", header, rows) - def handle_class_creation(self, value): + def handle_class_creation(self, value: List[Change]): # Create rows for the table rows = list({f"| {self._format_entity_labels(change.about_node)} |" for change in value}) @@ -235,7 +235,7 @@ def handle_class_creation(self, value): # Write the "Class Created" section as a markdown table self.write_markdown_table(f"Classes added: {len(rows)}", header, rows) - def handle_node_deletion(self, value): + def handle_node_deletion(self, value: List[Change]): # Create rows for the table rows = list({f"| {self._format_entity_labels(change.about_node)} |" for change in value}) @@ -245,7 +245,7 @@ def handle_node_deletion(self, value): # Write the "Nodes Deleted" section as a markdown table self.write_markdown_table(f"Nodes removed: {len(rows)}", header, rows) - def handle_new_text_definition(self, value): + def handle_new_text_definition(self, value: List[Change]): # Create rows for the table rows = list( { @@ -256,7 +256,7 @@ def handle_new_text_definition(self, value): header = "| Term | New Text Definition |" self.write_markdown_table(f"Text definitions added: {len(rows)}", header, rows) - def handle_remove_text_definition(self, value): + def handle_remove_text_definition(self, value: List[Change]): # Create rows for the table rows = list( { @@ -267,7 +267,7 @@ def handle_remove_text_definition(self, value): header = "| Term | Removed Text Definition |" self.write_markdown_table(f"Text definitions removed: {len(rows)}", header, rows) - def handle_node_obsoletion_with_direct_replacement(self, value): + def handle_node_obsoletion_with_direct_replacement(self, value: List[Change]): rows = list( { f"| {self._format_entity_labels(change.about_node)} |\ @@ -278,12 +278,12 @@ def handle_node_obsoletion_with_direct_replacement(self, value): header = "| Term | Replacement |" self.write_markdown_table(f"Nodes obsoleted with replacement: {len(rows)}", header, rows) - def handle_node_obsoletion(self, value): + def handle_node_obsoletion(self, value: List[Change]): rows = [f"| {self._format_entity_labels(change.about_node)} |" for change in value] header = "| Term |" self.write_markdown_table(f"Nodes obsoleted without replacement: {len(rows)}", header, rows) - def handle_node_direct_merge(self, value): + def handle_node_direct_merge(self, value: List[Change]): rows = list( { f"| {self._format_entity_labels(change.about_node)} |\ @@ -294,7 +294,7 @@ def handle_node_direct_merge(self, value): header = "| Term | Replacement |" self.write_markdown_table(f"Nodes merged: {len(rows)}", header, rows) - def handle_add_node_to_subset(self, value): + def handle_add_node_to_subset(self, value: List[Change]): rows = list( { f"| {self._format_entity_labels(change.about_node)} | {change.in_subset} |" @@ -304,7 +304,17 @@ def handle_add_node_to_subset(self, value): header = "| Term | Subset |" self.write_markdown_table(f"Nodes added to subset: {len(rows)}", header, rows) - def handle_mapping_creation(self, value): + def handle_remove_node_from_subset(self, value: List[Change]): + rows = list( + { + f"| {self._format_entity_labels(change.about_node)} | {change.in_subset} |" + for change in value + } + ) + header = "| Term | Subset |" + self.write_markdown_table(f"Nodes removed from subset: {len(rows)}", header, rows) + + def handle_mapping_creation(self, value: List[Change]): rows = list( { f"""| {self._format_entity_labels(change.subject)} | """ @@ -317,7 +327,7 @@ def handle_mapping_creation(self, value): header = "| Subject | Predicate | Object |" self.write_markdown_table(f"Mappings added: {len(rows)}", header, rows) - def handle_mapping_predicate_change(self, value): + def handle_mapping_predicate_change(self, value: List[Change]): # Create rows for the table rows = list( { @@ -335,7 +345,7 @@ def handle_mapping_predicate_change(self, value): # Write the "Predicate Changed" section as a collapsible markdown table self.write_markdown_table(f"Mappings changed: {len(rows)}", header, rows) - def handle_remove_mapping(self, value): + def handle_remove_mapping(self, value: List[Change]): rows = list( { f"""| {self._format_entity_labels(change.about_node)} | """ @@ -348,47 +358,47 @@ def handle_remove_mapping(self, value): header = "| Subject | Predicate | Object |" self.write_markdown_table(f"Mappings removed: {len(rows)}", header, rows) - # def handle_datatype_or_language_tag_change(self, value): + # def handle_datatype_or_language_tag_change(self, value:List[Change]): # # Implement datatype or language tag change handling logic here # logging.info("Datatype or language tag change handling not yet implemented.") - # def handle_language_tag_change(self, value): + # def handle_language_tag_change(self, value:List[Change]): # # Implement language tag change handling logic here # logging.info("Language tag change handling not yet implemented.") - # def handle_datatype_change(self, value): + # def handle_datatype_change(self, value:List[Change]): # # Implement datatype change handling logic here # logging.info("Datatype change handling not yet implemented.") - # def handle_allows_automatic_replacement_of_edges(self, value): + # def handle_allows_automatic_replacement_of_edges(self, value:List[Change]): # # Implement allows automatic replacement of edges handling logic here # logging.info("Allows automatic replacement of edges handling not yet implemented.") - # def handle_unobsoletion(self, value): + # def handle_unobsoletion(self, value:List[Change]): # # Implement unobsoletion handling logic here # logging.info("Unobsoletion handling not yet implemented.") - # def handle_deletion(self, value): + # def handle_deletion(self, value:List[Change]): # # Implement deletion handling logic here # logging.info("Deletion handling not yet implemented.") - # def handle_creation(self, value): + # def handle_creation(self, value:List[Change]): # # Implement creation handling logic here # logging.info("Creation handling not yet implemented.") - # def handle_subset_membership_change(self, value): + # def handle_subset_membership_change(self, value:List[Change]): # # Implement subset membership change handling logic here # logging.info("Subset membership change handling not yet implemented.") - # def handle_add_to_subset(self, value): + # def handle_add_to_subset(self, value:List[Change]): # # Implement add to subset handling logic here # logging.info("Add to subset handling not yet implemented.") - # def handle_remove_from_subset(self, value): + # def handle_remove_from_subset(self, value:List[Change]): # # Implement remove from subset handling logic here # logging.info("Remove from subset handling not yet implemented.") - # def handle_place_under(self, value): + # def handle_place_under(self, value:List[Change]): # # Implement place under handling logic here # logging.info("Place under handling not yet implemented.") @@ -417,6 +427,7 @@ def process_changes(self, curie_or_change: Dict[str, Change]): "EdgeCreation": self.handle_edge_creation, "EdgeChange": self.handle_edge_change, "AddNodeToSubset": self.handle_add_node_to_subset, + "RemoveNodeFromSubset": self.handle_remove_node_from_subset, "MappingPredicateChange": self.handle_mapping_predicate_change, "MappingCreation": self.handle_mapping_creation, "RemoveMapping": self.handle_remove_mapping,