diff --git a/rock_paper_sand/wikidata.py b/rock_paper_sand/wikidata.py index d82a463..e6d4c5b 100644 --- a/rock_paper_sand/wikidata.py +++ b/rock_paper_sand/wikidata.py @@ -200,6 +200,16 @@ def item(self, qid: str) -> Any: self._item_by_qid[qid] = response.json()["entities"][qid] return self._item_by_qid[qid] + def sparql(self, query: str) -> Any: + """Returns results from a SPARQL query.""" + response = self._session.get( + "https://query.wikidata.org/sparql", + params=[("query", query)], + headers={"Accept": "application/sparql-results+json"}, + ) + response.raise_for_status() + return response.json()["results"]["bindings"] + def _release_status( item: Any, diff --git a/rock_paper_sand/wikidata_test.py b/rock_paper_sand/wikidata_test.py index 46f2be4..ba8c261 100644 --- a/rock_paper_sand/wikidata_test.py +++ b/rock_paper_sand/wikidata_test.py @@ -105,6 +105,27 @@ def test_item(self) -> None: self._mock_session.mock_calls, ) + def test_sparql(self) -> None: + self._mock_session.get.return_value.json.return_value = { + "results": {"bindings": [{"foo": "bar"}]} + } + + results = self._api.sparql("SELECT ...") + + self.assertEqual([{"foo": "bar"}], results) + self.assertSequenceEqual( + ( + mock.call.get( + "https://query.wikidata.org/sparql", + params=[("query", "SELECT ...")], + headers={"Accept": "application/sparql-results+json"}, + ), + mock.call.get().raise_for_status(), + mock.call.get().json(), + ), + self._mock_session.mock_calls, + ) + class WikidataUtilsTest(parameterized.TestCase): # pylint: disable=protected-access