feat: implement query sanity checking

Closes #116, closes #126.

rdfproxy/adapter.py

@@ -5,6 +5,7 @@
 from typing import Generic
 
 from SPARQLWrapper import JSON, SPARQLWrapper
+from rdfproxy.checks.checkers import check_query
 from rdfproxy.mapper import ModelBindingsMapper
 from rdfproxy.utils._types import _TModelInstance
 from rdfproxy.utils.models import Page
@@ -34,7 +35,7 @@ class SPARQLModelAdapter(Generic[_TModelInstance]):
     def __init__(
         self, target: str | SPARQLWrapper, query: str, model: type[_TModelInstance]
     ) -> None:
-        self._query = query
+        self._query = check_query(query)
         self._model = model
 
         self.sparql_wrapper: SPARQLWrapper = (

rdfproxy/checks/checkers.py

@@ -0,0 +1,34 @@
+"""RDFProxy check runners."""
+
+from collections.abc import Callable, Iterable
+from typing import Annotated, NoReturn, TypeVar
+
+from rdfproxy.checks.query_checks import (
+    check_parse_query,
+    check_select_query,
+    check_solution_modifiers,
+)
+
+
+T = TypeVar("T")
+
+_TChecks = Iterable[Callable[[T], None | NoReturn]]
+
+
+def _check_factory(checks: _TChecks[T]) -> Callable[[T], T | NoReturn]:
+    """Produce an identity function that runs checks on its argument before returning."""
+
+    def _check(obj: T) -> T | NoReturn:
+        for check in checks:
+            check(obj)
+        return obj
+
+    return _check
+
+
+check_query: Annotated[
+    Callable[[str], str | NoReturn],
+    "Run query checks and return the query unless an exception is raised.",
+] = _check_factory(
+    checks=(check_parse_query, check_solution_modifiers, check_select_query)
+)

rdfproxy/checks/query_checks.py

@@ -0,0 +1,24 @@
+"""Query checks definitions."""
+
+from typing import NoReturn
+
+from rdflib.plugins.sparql.parser import parseQuery
+from rdfproxy.utils._exceptions import UnsupportedQueryException
+from rdfproxy.utils.predicates import (
+    query_has_solution_modifiers,
+    query_is_select_query,
+)
+
+
+def check_parse_query(query: str) -> None | NoReturn:
+    parseQuery(query)
+
+
+def check_select_query(query: str) -> None | NoReturn:
+    if not query_is_select_query(query):
+        raise UnsupportedQueryException("Only SELECT queries are applicable.")
+
+
+def check_solution_modifiers(query: str) -> None | NoReturn:
+    if query_has_solution_modifiers(query):
+        raise UnsupportedQueryException("SPARQL solution modifieres are not supported.")

rdfproxy/utils/_exceptions.py

@@ -15,3 +15,7 @@ class MissingModelConfigException(Exception):
 
 class UnboundGroupingKeyException(Exception):
     """Exception for indicating that no SPARQL binding corresponds to the requested grouping key."""
+
+
+class UnsupportedQueryException(Exception):
+    """Exception for indicating that a given SPARQL query is not supported."""

rdfproxy/utils/predicates.py

@@ -0,0 +1,18 @@
+"""RDFProxy predicate functions."""
+
+import re
+
+from rdflib.plugins.sparql.parser import parseQuery
+
+
+def query_is_select_query(query: str) -> bool:
+    """Check if a SPARQL query is a SELECT query."""
+    _, query_type = parseQuery(query)
+    return query_type.name == "SelectQuery"
+
+
+def query_has_solution_modifiers(query: str) -> bool:
+    """Predicate for checking if a SPARQL query has a solution modifier."""
+    pattern = r"}[^}]*\w+$"
+    result = re.search(pattern, query)
+    return bool(result)