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

@@ -15,6 +15,8 @@
 from pkg_api.core.namespaces import PKGPrefixes
 from pkg_api.core.pkg_types import URI, SPARQLQuery
 
+_SPARQL_STATEMENT_VARIABLE = "?statement"
+
 
 def _clean_sparql_representation(sparql: str) -> str:
     """Cleans a SPARQL representation.
@@ -222,7 +224,7 @@ def get_query_for_add_preference(pkg_data: PKGData) -> SPARQLQuery:
     query = f"""
         INSERT {{
             ?subject wi:preference [
-                pav:derivedFrom ?statement ;
+                pav:derivedFrom {_SPARQL_STATEMENT_VARIABLE} ;
                 wi:topic ?object ; wo:weight [
                     wo:weight_value "{pkg_data.preference.weight}"^^xsd:decimal;
                     wo:scale pkg:StandardScale
@@ -232,7 +234,7 @@ def get_query_for_add_preference(pkg_data: PKGData) -> SPARQLQuery:
         WHERE {{
             {statement_node_id} a rdf:Statement ;
             rdf:subject ?subject; rdf:object ?object .
-            BIND({statement_node_id} AS ?statement)
+            BIND({statement_node_id} AS {_SPARQL_STATEMENT_VARIABLE})
         }}
     """
 
@@ -258,10 +260,10 @@ def get_query_for_get_statements(pkg_data: PKGData) -> SPARQLQuery:
         SPARQL query.
     """
     statement_representation = _get_statement_representation(
-        pkg_data, "?statement"
+        pkg_data, _SPARQL_STATEMENT_VARIABLE
     )
     query = f"""
-        SELECT ?statement
+        SELECT {_SPARQL_STATEMENT_VARIABLE}
         WHERE {{
             {statement_representation}
         }}
@@ -295,14 +297,113 @@ def get_query_for_conditional_get_statements(triple: Triple) -> SPARQLQuery:
         if not annotation:
             continue
         value = _get_property_representation(annotation)
-        conditions.append(f"?statement {property} {value} .")
+        conditions.append(f"{_SPARQL_STATEMENT_VARIABLE} {property} {value} .")
 
     query = f"""
-        SELECT ?statement
+        SELECT {_SPARQL_STATEMENT_VARIABLE}
         WHERE {{
             {" ".join(conditions)}
         }}
     """
 
     # 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.
+    """
+    statement = _get_statement_representation(
+        pkg_data, _SPARQL_STATEMENT_VARIABLE
+    )
+    # 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 {_SPARQL_STATEMENT_VARIABLE} .
+            ?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.
+    """
+    statement_representation = _get_statement_representation(
+        pkg_data, _SPARQL_STATEMENT_VARIABLE
+    )
+    # Remove statement description from conditions
+    statement_representation = re.sub(
+        r'dc:description "[^"]+" ;', "", statement_representation
+    )
+
+    query = f"""
+        DELETE {{
+            {_SPARQL_STATEMENT_VARIABLE} ?p ?o .
+            ?preference ?pp ?op .
+        }}
+        WHERE {{
+            {statement_representation}
+            {_SPARQL_STATEMENT_VARIABLE} ?p ?o .
+            OPTIONAL {{
+                ?preference pav:derivedFrom {_SPARQL_STATEMENT_VARIABLE} .
+                ?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

@@ -1,6 +1,5 @@
 """Tests for utility methods."""
 
-
 import re
 import uuid
 from typing import Optional, Union
@@ -302,3 +301,68 @@ 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)
+
+
+def test_get_query_for_remove_preference(
+    pkg_data_example: PKGData, statement_representation: str
+) -> None:
+    """Tests get_query_for_remove_preference method."""
+    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 {{
+            ?preference ?p ?o .
+            ?subject wi:preference ?preference .
+        }}
+        WHERE {{
+            {statement_representation}
+            ?subject wi:preference ?preference .
+            ?preference pav:derivedFrom ?statement .
+            ?preference ?p ?o .
+        }}
+    """
+    assert utils.get_query_for_remove_preference(
+        pkg_data_example
+    ) == strip_string(sparql_query)