Use requests directly instead SPARQLWrapper

So tests can be closer to the real thing, by mocking the HTTP server
instead of having a completely different abstraction on the client side
(RdflibSparqlBackend)
This commit is contained in:
Val Lorentz 2023-05-21 19:57:52 +02:00
parent 370c75b16c
commit fb02fb3841
4 changed files with 26 additions and 27 deletions

View File

@ -17,7 +17,7 @@
import abc
from typing import Iterable
import SPARQLWrapper
import requests
class SparqlBackend(abc.ABC):
@ -37,9 +37,12 @@ class RemoteSparqlBackend(SparqlBackend):
:param agent: User-Agent to use in HTTP requests
"""
self._url = url
self._agent = agent
self._session = requests.Session()
self._session.headers["User-Agent"] = agent
def query(self, query: str) -> Iterable[tuple]:
sparql = SPARQLWrapper.SPARQLWrapper(self._url, agent=self._agent)
sparql.setQuery(query)
return sparql.query()
params = {"format": "json", "query": query}
resp = self._session.post(self._url, params=params).json()
variables = resp["head"]["vars"]
for result in resp["results"]["bindings"]:
yield tuple(result.get(variable) for variable in variables)

View File

@ -16,23 +16,10 @@
# pylint: disable=redefined-outer-name
from typing import Iterable
import pytest
import rdflib
from glowtables.sparql import SparqlBackend
class RdflibSparqlBackend(SparqlBackend):
"""Mock SPARQL backend that queries a local rdflib graph."""
def __init__(self, graph: rdflib.Graph):
self._graph = graph
def query(self, query: str) -> Iterable[tuple]:
ret = self._graph.query(query)
return ret
from glowtables.sparql import RemoteSparqlBackend
@pytest.fixture()
@ -42,6 +29,16 @@ def rdflib_graph() -> rdflib.Graph:
@pytest.fixture()
def rdflib_sparql(rdflib_graph: rdflib.Graph) -> RdflibSparqlBackend:
def rdflib_sparql(requests_mock, rdflib_graph: rdflib.Graph) -> RemoteSparqlBackend:
"""Returns a SPARQL backend instance for ``rdflib_graph``."""
return RdflibSparqlBackend(rdflib_graph)
def json_callback(request, context):
result = rdflib_graph.query(request.json())
context.status_code = 200
return {
"head": {"vars": result.vars},
"results": list(result),
}
requests_mock.register_uri("POST", "mock://sparql.example.org/", json=json_callback)
return RemoteSparqlBackend("mock://sparql.example.org/", agent="Mock Client")

View File

@ -17,12 +17,11 @@ import textwrap
import pytest
import rdflib
from glowtables.sparql import SparqlBackend
from glowtables.table import Language, LiteralField, Table
from .conftest import RdflibSparqlBackend
def test_single_literal(rdflib_sparql: RdflibSparqlBackend) -> None:
def test_single_literal(rdflib_sparql: SparqlBackend) -> None:
name_field = LiteralField(
"display_name",
{Language("en"): "Name"},
@ -46,7 +45,7 @@ def test_single_literal(rdflib_sparql: RdflibSparqlBackend) -> None:
rdflib_sparql.query(table.sparql())
def test_two_literals(rdflib_sparql: RdflibSparqlBackend) -> None:
def test_two_literals(rdflib_sparql: SparqlBackend) -> None:
name_field = LiteralField(
"display_name",
{Language("en"): "Name"},
@ -76,7 +75,7 @@ def test_two_literals(rdflib_sparql: RdflibSparqlBackend) -> None:
rdflib_sparql.query(table.sparql())
def test_default_value(rdflib_sparql: RdflibSparqlBackend) -> None:
def test_default_value(rdflib_sparql: SparqlBackend) -> None:
name_field = LiteralField(
"display_name",
{Language("en"): "Name"},

View File

@ -9,6 +9,7 @@ requires-python = ">=3.9"
dependencies = [
"flask ~= 2.0.0",
"rdflib ~= 6.0.0",
"requests ~= 3.0.0",
]
[project.optional-dependencies]
@ -29,7 +30,6 @@ python_version = "3.9"
[[tool.mypy.overrides]]
module = [
"requests_mock",
"SPARQLWrapper",
]
ignore_missing_imports = true