feat: query strategies

pyproject.toml

@@ -14,6 +14,7 @@ sparqlwrapper = "^2.0.0"
 pydantic = "^2.9.2"
 
 
+httpx = "^0.28.1"
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.7.0"
 deptry = "^0.20.0"

rdfproxy/__init__.py

@@ -1,4 +1,9 @@
 from rdfproxy.adapter import SPARQLModelAdapter  # noqa: F401
 from rdfproxy.mapper import ModelBindingsMapper  # noqa: F401
+from rdfproxy.sparql_strategies import (
+    SPARQLStrategy,  # noqa: F401
+    SPARQLWrapperStrategy,  # noqa: F401
+    HttpxStrategy,  # noqa: F401
+)
 from rdfproxy.utils._types import ConfigDict, SPARQLBinding  # noqa: F401
 from rdfproxy.utils.models import Page, QueryParameters  # noqa: F401

rdfproxy/adapter.py

@@ -4,15 +4,14 @@
 import math
 from typing import Generic
 
-from SPARQLWrapper import JSON, SPARQLWrapper
 from rdfproxy.mapper import ModelBindingsMapper
+from rdfproxy.sparql_strategies import HttpxStrategy, SPARQLStrategy
 from rdfproxy.utils._types import _TModelInstance
 from rdfproxy.utils.models import Page, QueryParameters
 from rdfproxy.utils.sparql_utils import (
     calculate_offset,
     construct_count_query,
     construct_items_query,
-    query_with_wrapper,
 )
 
 
@@ -32,15 +31,16 @@ class SPARQLModelAdapter(Generic[_TModelInstance]):
     """
 
     def __init__(
-        self, target: str | SPARQLWrapper, query: str, model: type[_TModelInstance]
+        self,
+        target: str,
+        query: str,
+        model: type[_TModelInstance],
+        sparql_strategy: type[SPARQLStrategy] = HttpxStrategy,
     ) -> None:
         self._query = query
         self._model = model
 
-        self.sparql_wrapper: SPARQLWrapper = (
-            SPARQLWrapper(target) if isinstance(target, str) else target
-        )
-        self.sparql_wrapper.setReturnFormat(JSON)
+        self.sparql_strategy = sparql_strategy(target)
 
     def query(self, query_parameters: QueryParameters) -> Page[_TModelInstance]:
         """Run a query against an endpoint and return a Page model object."""
@@ -52,9 +52,7 @@ def query(self, query_parameters: QueryParameters) -> Page[_TModelInstance]:
             offset=calculate_offset(query_parameters.page, query_parameters.size),
         )
 
-        items_query_bindings: Iterator[dict] = query_with_wrapper(
-            query=items_query, sparql_wrapper=self.sparql_wrapper
-        )
+        items_query_bindings: Iterator[dict] = self.sparql_strategy.query(items_query)
 
         mapper = ModelBindingsMapper(self._model, *items_query_bindings)
 
@@ -75,7 +73,5 @@ def _get_count(self, query: str) -> int:
 
         Helper for SPARQLModelAdapter.query.
         """
-        result: Iterator[dict] = query_with_wrapper(
-            query=query, sparql_wrapper=self.sparql_wrapper
-        )
+        result: Iterator[dict] = self.sparql_strategy.query(query)
         return int(next(result)["cnt"])

rdfproxy/sparql_strategies.py

@@ -0,0 +1,58 @@
+"""Strategy classes for SPARQL query functionality."""
+
+import abc
+from collections.abc import Iterator
+
+from SPARQLWrapper import JSON, QueryResult, SPARQLWrapper
+import httpx
+
+
+class SPARQLStrategy(abc.ABC):
+    def __init__(self, endpoint: str):
+        self.endpoint = endpoint
+
+    @abc.abstractmethod
+    def query(self, sparql_query: str) -> Iterator[dict[str, str]]:
+        raise NotImplementedError
+
+    @staticmethod
+    def _get_bindings_from_bindings_dict(bindings_dict: dict) -> Iterator[dict]:
+        bindings = map(
+            lambda binding: {k: v["value"] for k, v in binding.items()},
+            bindings_dict["results"]["bindings"],
+        )
+        return bindings
+
+
+class SPARQLWrapperStrategy(SPARQLStrategy):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self._sparql_wrapper = SPARQLWrapper(self.endpoint)
+        self._sparql_wrapper.setReturnFormat(JSON)
+
+    def query(self, sparql_query: str) -> Iterator[dict[str, str]]:
+        self._sparql_wrapper.setQuery(sparql_query)
+
+        result: QueryResult = self._sparql_wrapper.query()
+        return self._get_bindings_from_bindings_dict(result.convert())
+
+
+class HttpxStrategy(SPARQLStrategy):
+    def query(self, sparql_query: str) -> Iterator[dict[str, str]]:
+        result: httpx.Response = self._httpx_run_sparql_query(sparql_query)
+        return self._get_bindings_from_bindings_dict(result.json())
+
+    def _httpx_run_sparql_query(self, query: str) -> httpx.Response:
+        data = {"output": "json", "query": query}
+        headers = {
+            "Accept": "application/sparql-results+json",
+        }
+
+        response = httpx.post(
+            self.endpoint,
+            headers=headers,
+            data=data,
+        )
+
+        return response