Add query to delete statement

pkg_api/pkg.py

@@ -347,3 +347,19 @@ def _retrieve_and_parse_concept(
             return None
 
         return Concept(**concept_dict)
+
+    def remove_statement(self, pkg_data: PKGData) -> None:
+        """Removes a statement from the PKG.
+
+        Args:
+            pkg_data: PKG data associated to the statement.
+        """
+        # Remove preference derived from the statement, if any
+        query = utils.get_query_for_remove_preference(pkg_data)
+        self._connector.execute_sparql_update(query)
+        # Remove statement
+        query = utils.get_query_for_remove_statement(pkg_data)
+        self._connector.execute_sparql_update(query)
+        # Remove dangling concepts and scales
+        for query in utils.get_queries_for_remove_cleanup():
+            self._connector.execute_sparql_update(query)

pkg_api/utils.py

@@ -306,3 +306,102 @@ def get_query_for_conditional_get_statements(triple: Triple) -> SPARQLQuery:
 
     # Cleaning up the query
     return _clean_sparql_representation(query)
+
+
+def get_query_for_remove_preference(pkg_data: PKGData) -> SPARQLQuery:
+    """Gets SPARQL query to remove a preference.
+
+    Args:
+        pkg_data: PKG data associated to a statement.
+
+    Returns:
+        SPARQL query.
+    """
+    blank_node_id = "?statement"
+    statement = _get_statement_representation(pkg_data, blank_node_id)
+    # Remove statement description from conditions
+    statement = re.sub(r'dc:description "[^"]+" ;', "", statement)
+
+    query = f"""
+        DELETE {{
+            ?preference ?p ?o .
+            ?subject wi:preference ?preference .
+        }}
+        WHERE {{
+            {statement}
+            ?subject wi:preference ?preference .
+            ?preference pav:derivedFrom {blank_node_id} .
+            ?preference ?p ?o .
+        }}
+    """
+
+    # Cleaning up the query
+    return _clean_sparql_representation(query)
+
+
+def get_query_for_remove_statement(pkg_data: PKGData) -> SPARQLQuery:
+    """Gets SPARQL query to remove a statement.
+
+    Note that if a preference is derived from the statement, it is also removed.
+
+    Args:
+        pkg_data: PKG data associated to a statement.
+
+    Returns:
+        SPARQL query.
+    """
+    blank_node_id = "?statement"
+    statement_representation = _get_statement_representation(
+        pkg_data, blank_node_id
+    )
+    # Remove statement description from conditions
+    statement_representation = re.sub(
+        r'dc:description "[^"]+" ;', "", statement_representation
+    )
+
+    query = f"""
+        DELETE {{
+            ?statement ?p ?o .
+            ?preference ?pp ?op .
+        }}
+        WHERE {{
+            {statement_representation}
+            ?statement ?p ?o .
+            OPTIONAL {{
+                ?preference pav:derivedFrom ?statement .
+                ?preference ?pp ?op .
+            }}
+        }}
+    """
+
+    # Cleaning up the query
+    return _clean_sparql_representation(query)
+
+
+def get_queries_for_remove_cleanup() -> List[SPARQLQuery]:
+    """Gets SPARQL queries to delete dangling concepts and weight scales.
+
+    Returns:
+        List of SPARQL queries.
+    """
+    query_delete_concepts = """
+    DELETE {
+        ?concept ?p ?o .
+    } WHERE {
+        ?concept a skos:Concept .
+        ?concept ?p ?o .
+        FILTER NOT EXISTS { ?_1 ?_2 ?concept . }
+    }
+    """
+    query_delete_concepts = _clean_sparql_representation(query_delete_concepts)
+    query_delete_scales = """
+    DELETE {
+        ?x ?p ?o .
+    } WHERE {
+        ?x wo:scale ?_3 .
+        ?x ?p ?o .
+        FILTER NOT EXISTS { ?_1 ?_2 ?x . }
+    }
+    """
+    query_delete_scales = _clean_sparql_representation(query_delete_scales)
+    return [query_delete_concepts, query_delete_scales]

tests/pkg_api/test_pkg.py

@@ -179,3 +179,14 @@ def test_get_statements_with_triple_conditions(
     )
     assert len(statements) == 2
     assert retrieved_statement_with_concept in statements
+
+
+def test_remove_statement(user_pkg: PKG, statement: PKGData) -> None:
+    """Tests removing a statement."""
+    user_pkg.add_statement(statement)
+    statements = user_pkg.get_statements(statement)
+    assert len(statements) == 1
+
+    user_pkg.remove_statement(statement)
+    statements = user_pkg.get_statements(statement)
+    assert len(statements) == 0

tests/pkg_api/test_utils.py

@@ -302,3 +302,39 @@ def test_get_query_for_conditional_get_statements(
     assert utils.get_query_for_conditional_get_statements(
         pkg_data_example.triple
     ) == strip_string(expected_query)
+
+
+def test_get_query_for_remove_statement(
+    pkg_data_example: PKGData, statement_representation: str
+) -> None:
+    """Tests get_query_for_remove_statement method.
+
+    Args:
+        pkg_data_example: PKG data example.
+        statement_representation: Statement representation.
+    """
+    statement_node_id = utils.get_statement_node_id(pkg_data_example)
+    statement_representation = statement_representation.replace(
+        statement_node_id, "?statement"
+    )
+    statement_representation = re.sub(
+        r'dc:description "[^"]+" ;', "", statement_representation
+    )
+    sparql_query = f"""
+        DELETE {{
+            ?statement ?p ?o .
+            ?preference ?pp ?op .
+        }}
+        WHERE {{
+            {statement_representation}
+            ?statement ?p ?o .
+            OPTIONAL {{
+                ?preference pav:derivedFrom ?statement .
+                ?preference ?pp ?op .
+            }}
+        }}
+    """
+
+    assert utils.get_query_for_remove_statement(
+        pkg_data_example
+    ) == strip_string(sparql_query)