Add query to SPARQLwrapper strategy

tests/backends/test_sparqlwrapper.py

@@ -0,0 +1,29 @@
+"""Test SPARQLwrapper backend."""
+
+# pylint: disable=invalid-name
+
+
+def test_sparqlwrapper_backend():
+    """Test SPARQLwrapper backend."""
+    import pytest
+
+    pytest.importorskip("SPARQLWrapper")
+
+    from tripper import Triplestore
+
+    # Requires internet connection to connect to wikidata
+    sparql_query = """
+#Country where the capital is Oslo.
+SELECT DISTINCT ?country ?countryLabel
+WHERE
+{
+  ?country wdt:P31 wd:Q3624078 . # isA country
+  ?country wdt:P36 wd:Q585 . # captial Oslo
+  SERVICE wikibase:label { bd:serviceParam wikibase:language "en" }
+}
+"""
+
+    endpoint_url = "https://query.wikidata.org/sparql"
+    ts = Triplestore(backend="sparqlwrapper", base_iri=endpoint_url)
+    res = ts.query(sparql_query)
+    assert res == [("http://www.wikidata.org/entity/Q20", "Norway")]

tripper/backends/sparqlwrapper.py

@@ -16,7 +16,7 @@
 
 if TYPE_CHECKING:  # pragma: no cover
     from collections.abc import Sequence
-    from typing import Dict, Generator
+    from typing import Dict, Generator, List, Tuple, Union
 
     from SPARQLWrapper import QueryResult
     from triplestore import Triple
@@ -39,6 +39,31 @@ def __init__(self, base_iri: str, **kwargs) -> None:
         )  # database is not used in the SPARQLWrapper backend
         self.sparql = SPARQLWrapper(endpoint=base_iri, **kwargs)
 
+    def query(
+        self, query_object, **kwargs  # pylint: disable=unused-argument
+    ) -> "Union[List[Tuple[str, ...]], bool, Generator[Triple, None, None]]":
+        """SPARQL query.
+
+        Parameters:
+            query_object: String with the SPARQL query.
+            kwargs: Keyword arguments passed to rdflib.Graph.query().
+
+        Returns:
+            The return type depends on type of query:
+              - SELECT: list of tuples of IRIs for each matching row
+              - ASK: TODO
+              - CONSTRUCT, DESCRIBE: TODO
+        """
+        self.sparql.setReturnFormat(JSON)
+        self.sparql.setMethod(POST)
+        self.sparql.setQuery(query_object)
+        ret = self.sparql.queryAndConvert()
+        bindings = ret["results"]["bindings"]
+        return [
+            tuple(str(convert_json_entrydict(v)) for v in row.values())
+            for row in bindings
+        ]
+
     def triples(self, triple: "Triple") -> "Generator[Triple, None, None]":
         """Returns a generator over matching triples."""
         variables = [