Skip to content

Commit

Permalink
feat: implement RDFProxy-compliance checkers for models
Browse files Browse the repository at this point in the history
Closes #108.
  • Loading branch information
lu-pl committed Feb 14, 2025
1 parent a7cd598 commit e2941e5
Showing 1 changed file with 79 additions and 0 deletions.
79 changes: 79 additions & 0 deletions rdfproxy/utils/checkers/model_checker.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
"""Functionality for performing RDFProxy-compliance checks on Pydantic models."""

from typing import TypeVar

from pydantic import BaseModel
from rdfproxy.utils.mapper_utils import _is_list_type
from rdfproxy.utils.utils import compose_left


_TModel = TypeVar("_TModel", bound=BaseModel)


class RDFProxyModelValidationException(Exception):
"""Exception for indicating that a model is invalid according to RDFProxy semantics"""


class RDFProxyGroupByException(RDFProxyModelValidationException):
"""Exception for indicating invalid group_by definitions."""


class RDFProxyModelBoolException(RDFProxyModelValidationException):
"""Exception for indicating invalid model_bool definitions."""


def _check_group_by_config(model: type[_TModel]) -> type[_TModel]:
"""Model checker for group_by config settings and grouping model semantics."""
model_group_by_value: str | None = model.model_config.get("group_by")
model_has_list_field: bool = any(
_is_list_type(value.annotation) for value in model.model_fields.values()
)

match model_group_by_value, model_has_list_field:
case None, False:
return model

case None, True:
raise RDFProxyGroupByException(
f"Model '{model.__name__}' has a list-annotated field "
"but does not specify 'group_by' in its model_config."
)

case str(), False:
raise RDFProxyGroupByException(
f"Model '{model.__name__}' does not specify "
"a grouping target (i.e. a list-annotated field)."
)

case str(), True:
applicable_keys: list[str] = [
k
for k, v in model.model_fields.items()
if not _is_list_type(v.annotation)
]

if model_group_by_value in applicable_keys:
return model

applicable_fields_message: str = (
"No applicable fields."
if not applicable_keys
else f"Applicable grouping field(s): {', '.join(applicable_keys)}"
)

raise RDFProxyGroupByException(
f"Requested grouping key '{model_group_by_value}' does not denote "
f"an applicable grouping field. {applicable_fields_message}"
)

case _: # pragma: no cover
raise AssertionError("This should never happen.")


def _check_model_bool_config(model: type[_TModel]) -> type[_TModel]:
"""Model checker for model_bool config settings."""
return model


def check_model(model: type[_TModel]) -> type[_TModel]:
return compose_left(_check_group_by_config, _check_model_bool_config)(model)

0 comments on commit e2941e5

Please sign in to comment.