Skip to content

Commit

Permalink
refactor: remove obsolete functions from mapper_utils
Browse files Browse the repository at this point in the history
  • Loading branch information
lu-pl committed Jan 22, 2025
1 parent 6713b91 commit 933da4a
Showing 1 changed file with 3 additions and 64 deletions.
67 changes: 3 additions & 64 deletions rdfproxy/utils/mapper_utils.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,8 @@
from collections.abc import Callable, Iterable
from typing import Any, TypeGuard, get_args, get_origin
from collections.abc import Iterable
from typing import TypeGuard, get_args, get_origin

from pydantic import BaseModel
from pydantic.fields import FieldInfo
from rdfproxy.utils._exceptions import (
InvalidGroupingKeyException,
MissingModelConfigException,
)
from rdfproxy.utils._types import _TModelInstance
from rdfproxy.utils._types import ModelBoolPredicate, SPARQLBinding, _TModelBoolValue
from rdfproxy.utils._types import ModelBoolPredicate, _TModelBoolValue


def _is_type(obj: type | None, _type: type) -> bool:
Expand All @@ -28,61 +22,6 @@ def _is_list_basemodel_type(obj: type | None) -> bool:
)


def _collect_values_from_bindings(
binding_name: str,
bindings: Iterable[dict],
predicate: Callable[[Any], bool] = lambda x: x is not None,
) -> list:
"""Scan bindings for a key binding_name and collect unique predicate-compliant values.
Note that element order is important for testing, so a set cast won't do.
"""
values = dict.fromkeys(
value
for binding in bindings
if predicate(value := binding.get(binding_name, None))
)
return list(values)


def _get_key_from_metadata(v: FieldInfo, *, default: Any) -> str | Any:
"""Try to get a SPARQLBinding object from a field's metadata attribute.
Helper for _generate_binding_pairs.
"""
return next(filter(lambda x: isinstance(x, SPARQLBinding), v.metadata), default)


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
except KeyError as e:
raise MissingModelConfigException(
"Model config with 'group_by' value required "
"for field-based grouping behavior."
) from e
else:
applicable_keys = _get_applicable_grouping_keys(model=model)

if group_by not in applicable_keys:
raise InvalidGroupingKeyException(
f"Invalid grouping key '{group_by}'. "
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


def default_model_bool_predicate(model: BaseModel) -> bool:
"""Default predicate for determining model truthiness.
Expand Down

0 comments on commit 933da4a

Please sign in to comment.