Skip to content

Commit

Permalink
feat: implement grouping by model field
Browse files Browse the repository at this point in the history
Currently, the group_by model_config parameter expects the name of a
SPARQLBinding; the rationale for this behavior was, that grouping
should theoretically also be possible for bindings that are not part
of the model.

Grouping is however primarily concerded with the model i.e. shapes
definition, grouping by a model field is therefore more intuitive and
conceptually more consistent.

Grouping by bindings should not be part of the model is still possible by using
pydantic.Field(exclude=True). This also forces backend implementers to
be explicit about 'external' grouping keys.

Closes #146.
  • Loading branch information
lu-pl committed Dec 2, 2024
1 parent ebf5e0d commit cfd3f83
Show file tree
Hide file tree
Showing 3 changed files with 25 additions and 12 deletions.
8 changes: 4 additions & 4 deletions rdfproxy/mapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,9 +40,9 @@ def _get_unique_models(self, model, bindings):

return models

def _get_group_by(self, model, kwargs) -> str:
def _get_group_by(self, model) -> str:
"""Get the group_by value from a model and register it in self._contexts."""
group_by: str = _get_group_by(model, kwargs)
group_by: str = _get_group_by(model)

if group_by not in self._contexts:
self._contexts.append(group_by)
Expand All @@ -57,7 +57,7 @@ def _generate_binding_pairs(
"""Generate an Iterator[tuple] projection of the bindings needed for model instantation."""
for k, v in model.model_fields.items():
if _is_list_basemodel_type(v.annotation):
group_by: str = self._get_group_by(model, kwargs)
group_by: str = self._get_group_by(model)
group_model, *_ = get_args(v.annotation)

applicable_bindings = filter(
Expand All @@ -69,7 +69,7 @@ def _generate_binding_pairs(
value = self._get_unique_models(group_model, applicable_bindings)

elif _is_list_type(v.annotation):
group_by: str = self._get_group_by(model, kwargs)
group_by: str = self._get_group_by(model)
applicable_bindings = filter(
lambda x: x[group_by] == kwargs[group_by],
self.bindings,
Expand Down
4 changes: 2 additions & 2 deletions rdfproxy/utils/_exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@ class MissingModelConfigException(Exception):
"""Exception for indicating that an expected Config class is missing in a Pydantic model definition."""


class UnboundGroupingKeyException(Exception):
"""Exception for indicating that no SPARQL binding corresponds to the requested grouping key."""
class InvalidGroupingKeyException(Exception):
"""Exception for indicating that an invalid grouping key has been encountered."""


class QueryConstructionException(Exception):
Expand Down
25 changes: 19 additions & 6 deletions rdfproxy/utils/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,10 @@
from pydantic import BaseModel
from pydantic.fields import FieldInfo
from rdfproxy.utils._exceptions import (
InvalidGroupingKeyException,
MissingModelConfigException,
UnboundGroupingKeyException,
)
from rdfproxy.utils._types import _TModelInstance
from rdfproxy.utils._types import ModelBoolPredicate, SPARQLBinding, _TModelBoolValue


Expand Down Expand Up @@ -54,7 +55,11 @@ def _get_key_from_metadata(v: FieldInfo, *, default: Any) -> str | Any:
return next(filter(lambda x: isinstance(x, SPARQLBinding), v.metadata), default)


def _get_group_by(model: type[BaseModel], kwargs: dict) -> str:
def _get_applicable_grouping_keys(model: type[_TModelInstance]) -> list[str]:
return [k for k, v in model.model_fields.items() if not _is_list_type(v.annotation)]


def _get_group_by(model: type[_TModelInstance]) -> str:
"""Get the name of a grouping key from a model Config class."""
try:
group_by = model.model_config["group_by"] # type: ignore
Expand All @@ -64,11 +69,19 @@ def _get_group_by(model: type[BaseModel], kwargs: dict) -> str:
"for field-based grouping behavior."
) from e
else:
if group_by not in kwargs.keys():
raise UnboundGroupingKeyException(
f"Requested grouping key '{group_by}' not in SPARQL binding projection.\n"
f"Applicable grouping keys: {', '.join(kwargs.keys())}."
applicable_keys = _get_applicable_grouping_keys(model=model)

if group_by not in applicable_keys:
raise InvalidGroupingKeyException(
f"Invalid grouping key '{group_by}'.\n"
f"Applicable grouping keys: {', '.join(applicable_keys)}."
)

if meta := model.model_fields[group_by].metadata:
if binding := next(
filter(lambda entry: isinstance(entry, SPARQLBinding), meta), None
):
return binding
return group_by


Expand Down

0 comments on commit cfd3f83

Please sign in to comment.