feat: implement query sanity checking

pyproject.toml

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

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
+from typing import Annotated, NoReturn, TypeVar
+
+from rdfproxy.checks.query_checks import (
+    check_parse_query,
+    check_select_query,
+    check_solution_modifiers,
+)
+
+
+T = TypeVar("T")
+_TCheck = Callable[[T], T | NoReturn]
+
+
+def compose_left(*fns: Callable) -> Callable:
+    def _left_wrapper(*fns):
+        fn, *rest_fns = fns
+
+        if rest_fns:
+            return lambda *args, **kwargs: fn(_left_wrapper(*rest_fns)(*args, **kwargs))
+        return fn
+
+    return _left_wrapper(*reversed(fns))
+
+
+def compose_checker(*checkers: _TCheck) -> _TCheck:
+    return compose_left(*checkers)
+
+
+check_query: Annotated[
+    _TCheck, "Run a series of checks on a query and return the query."
+] = compose_checker(check_parse_query, check_select_query, check_solution_modifiers)

rdfproxy/checks/query_checks.py

@@ -0,0 +1,29 @@
+"""Query checks definitions."""
+
+from typing import NoReturn, TypeVar
+
+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,
+)
+
+TQuery = TypeVar("TQuery", bound=str)
+
+
+def check_parse_query(query: TQuery) -> TQuery | NoReturn:
+    parseQuery(query)
+    return query
+
+
+def check_select_query(query: TQuery) -> TQuery | NoReturn:
+    if not query_is_select_query(query):
+        raise UnsupportedQueryException("Only SELECT queries are applicable.")
+    return query
+
+
+def check_solution_modifiers(query: TQuery) -> TQuery | NoReturn:
+    if query_has_solution_modifiers(query):
+        raise UnsupportedQueryException("SPARQL solution modifieres are not supported.")
+    return query

rdfproxy/utils/_exceptions.py

@@ -7,3 +7,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)