diff --git a/src/aind_slims_api/core.py b/src/aind_slims_api/core.py index 4542d41..119e010 100644 --- a/src/aind_slims_api/core.py +++ b/src/aind_slims_api/core.py @@ -10,18 +10,18 @@ import base64 import logging -from copy import deepcopy from functools import lru_cache -from typing import Any, Optional, Type, TypeVar, get_type_hints +from typing import Any, Optional, Type, TypeVar from pydantic import ValidationError from requests import Response -from slims.criteria import Criterion, Expression, Junction, conjunction, equals +from slims.criteria import Criterion, conjunction, equals from slims.internal import Record as SlimsRecord from slims.slims import Slims, _SlimsApiException from aind_slims_api import config from aind_slims_api.exceptions import SlimsRecordNotFound +from aind_slims_api.filters import resolve_filter_args from aind_slims_api.models import SlimsAttachment from aind_slims_api.models.base import SlimsBaseModel from aind_slims_api.types import SLIMS_TABLES @@ -104,35 +104,6 @@ def fetch( return records - @staticmethod - def resolve_model_alias( - model: Type[SlimsBaseModelTypeVar], - attr_name: str, - ) -> str: - """Given a SlimsBaseModel object, resolve its pk to the actual value - - Notes - ----- - - Raises ValueError if the alias cannot be resolved - - Resolves the validation alias for a given field name - - If prefixed with `-` will return the validation alias prefixed with - `-` - """ - has_prefix = attr_name.startswith("-") - _attr_name = attr_name.lstrip("-") - for field_name, field_info in model.model_fields.items(): - if ( - field_name == _attr_name - and field_info.validation_alias - and isinstance(field_info.validation_alias, str) - ): - alias = field_info.validation_alias - if has_prefix: - return f"-{alias}" - return alias - else: - raise ValueError(f"Cannot resolve alias for {attr_name} on {model}") - @staticmethod def _validate_models( model_type: Type[SlimsBaseModelTypeVar], records: list[SlimsRecord] @@ -147,141 +118,6 @@ def _validate_models( logger.error(f"SLIMS data validation failed, {repr(e)}") return validated - @staticmethod - def _resolve_criteria( - model_type: Type[SlimsBaseModelTypeVar], criteria: Criterion - ) -> Criterion: - """Resolves criterion field name to serialization alias in a criterion.""" - if isinstance(criteria, Junction): - criteria.members = [ - SlimsClient._resolve_criteria(model_type, sub_criteria) - for sub_criteria in criteria.members - ] - return criteria - elif isinstance(criteria, Expression): - if criteria.criterion["fieldName"] == "isNaFilter": - criteria.criterion["value"] = SlimsClient.resolve_model_alias( - model_type, - criteria.criterion["value"], - ) - else: - criteria.criterion["fieldName"] = SlimsClient.resolve_model_alias( - model_type, - criteria.criterion["fieldName"], - ) - return criteria - else: - raise ValueError(f"Invalid criterion type: {type(criteria)}") - - @staticmethod - def _validate_field_name( - model_type: Type[SlimsBaseModelTypeVar], - field_name: str, - ) -> None: - """Check if field_name is a field on a model. Raises a ValueError if it - is not. - """ - field_type_map = get_type_hints(model_type) - if field_name not in field_type_map: - raise ValueError(f"{field_name} is not a field on {model_type}.") - - @staticmethod - def _validate_field_value( - model_type: Type[SlimsBaseModelTypeVar], - field_name: str, - field_value: Any, - ) -> None: - """Check if field_value is a compatible with - the type associated with that field. Raises a ValueError if it is not. - """ - field_type_map = get_type_hints(model_type) - field_type = field_type_map[field_name] - if not isinstance(field_value, field_type): - raise ValueError( - f"{field_value} is incompatible with {field_type}" - f" for field {field_name}" - ) - - @staticmethod - def _validate_criteria( - model_type: Type[SlimsBaseModelTypeVar], criteria: Criterion - ) -> None: - """Validates that the types used in a criterion are compatible with the - types on the model. Raises a ValueError if they are not. - """ - if isinstance(criteria, Junction): - for sub_criteria in criteria.members: - SlimsClient._validate_criteria(model_type, sub_criteria) - elif isinstance(criteria, Expression): - if criteria.criterion["fieldName"] == "isNaFilter": - SlimsClient._validate_field_name( - model_type, - criteria.criterion["value"], - ) - elif criteria.criterion["operator"] in ["inSet", "notInSet"]: - for value in criteria.criterion["value"]: - SlimsClient._validate_field_name( - model_type, - criteria.criterion["fieldName"], - ) - SlimsClient._validate_field_value( - model_type, - criteria.criterion["fieldName"], - value, - ) - elif criteria.criterion["operator"] == "betweenInclusive": - SlimsClient._validate_field_name( - model_type, - criteria.criterion["fieldName"], - ) - SlimsClient._validate_field_value( - model_type, - criteria.criterion["fieldName"], - criteria.criterion["start"], - ) - SlimsClient._validate_field_value( - model_type, - criteria.criterion["fieldName"], - criteria.criterion["end"], - ) - else: - SlimsClient._validate_field_name( - model_type, - criteria.criterion["fieldName"], - ) - SlimsClient._validate_field_value( - model_type, - criteria.criterion["fieldName"], - criteria.criterion["value"], - ) - else: - raise ValueError(f"Invalid criterion type: {type(criteria)}") - - @staticmethod - def _resolve_filter_args( - model: Type[SlimsBaseModelTypeVar], - *args: Criterion, - sort: list[str] = [], - start: Optional[int] = None, - end: Optional[int] = None, - **kwargs, - ) -> tuple[list[Criterion], list[str], Optional[int], Optional[int]]: - """Validates filter arguments and resolves field names to SLIMS API - column names. - """ - criteria = deepcopy(list(args)) - criteria.extend(map(lambda item: equals(item[0], item[1]), kwargs.items())) - resolved_criteria: list[Criterion] = [] - for criterion in criteria: - SlimsClient._validate_criteria(model, criterion) - resolved_criteria.append(SlimsClient._resolve_criteria(model, criterion)) - resolved_sort = [ - SlimsClient.resolve_model_alias(model, sort_key) for sort_key in sort - ] - if start is not None and end is None or end is not None and start is None: - raise ValueError("Must provide both start and end or neither for fetch.") - return resolved_criteria, resolved_sort, start, end - def fetch_models( self, model: Type[SlimsBaseModelTypeVar], @@ -307,7 +143,7 @@ def fetch_models( if isinstance(sort, str): sort = [sort] - criteria, resolved_sort, start, end = self._resolve_filter_args( + criteria, resolved_sort, start, end = resolve_filter_args( model, *args, sort=sort, @@ -391,7 +227,7 @@ def fetch_attachments( if isinstance(sort, str): sort = [sort] - criteria, sort, start, end = self._resolve_filter_args( + criteria, sort, start, end = resolve_filter_args( SlimsAttachment, *args, sort=sort, diff --git a/src/aind_slims_api/filters.py b/src/aind_slims_api/filters.py index 33b9004..65fd455 100644 --- a/src/aind_slims_api/filters.py +++ b/src/aind_slims_api/filters.py @@ -6,6 +6,7 @@ from typing import Any, Optional, Type, TypeVar, get_type_hints from slims.criteria import Criterion, Expression, Junction, equals + from aind_slims_api.models.base import SlimsBaseModel logger = logging.getLogger(__name__) @@ -17,7 +18,8 @@ def resolve_model_alias( model: Type[SlimsBaseModelTypeVar], attr_name: str, ) -> str: - """Given a SlimsBaseModel object, resolve its pk to the actual value + """Given a SlimsBaseModel object, resolve an alias name to the actual slims + column name. Notes ----- @@ -42,19 +44,22 @@ def resolve_model_alias( raise ValueError(f"Cannot resolve alias for {attr_name} on {model}") -def _resolve_criteria( +CriteriaType = TypeVar("CriteriaType", Junction, Expression) + + +def resolve_criteria( model_type: Type[SlimsBaseModelTypeVar], - criteria: Criterion, -) -> Criterion: + criteria: Type[CriteriaType], +) -> Type[CriteriaType]: """Resolves criterion field name to serialization alias in a criterion.""" if isinstance(criteria, Junction): criteria.members = [ - _resolve_criteria(model_type, sub_criteria) + resolve_criteria(model_type, sub_criteria) for sub_criteria in criteria.members ] return criteria - elif isinstance(criteria, Expression): + else: # Expression if criteria.criterion["fieldName"] == "isNaFilter": criteria.criterion["value"] = resolve_model_alias( model_type, @@ -66,29 +71,14 @@ def _resolve_criteria( criteria.criterion["fieldName"], ) return criteria - else: - raise ValueError(f"Invalid criterion type: {type(criteria)}") - - -def validate_criterion( - model_type: Type[SlimsBaseModelTypeVar], - field_name: str, -) -> None: - """Check if field_name is a field on a model. Raises a ValueError if it - is not. - """ - field_type_map = get_type_hints(model_type) - if field_name not in field_type_map: - raise ValueError(f"{field_name} is not a field on {model_type}.") - def _validate_field_name( model_type: Type[SlimsBaseModelTypeVar], field_name: str, ) -> None: - """Check if field_name is a field on a model. Raises a ValueError if it - is not. + """Check if field_name is a field on a model. Raises an Attribute error if + it is not. """ field_type_map = get_type_hints(model_type) if field_name not in field_type_map: @@ -100,8 +90,8 @@ def _validate_field_value( field_name: str, field_value: Any, ) -> None: - """Check if field_value is a compatible with - the type associated with that field. Raises a ValueError if it is not. + """Check if field_value is a compatible with the type associated with that + field. Raises a ValueError if it is not. """ field_type_map = get_type_hints(model_type) field_type = field_type_map[field_name] @@ -112,11 +102,18 @@ def _validate_field_value( ) -def _validate_criteria( - model_type: Type[SlimsBaseModelTypeVar], criteria: Criterion +def validate_criteria( + model_type: Type[SlimsBaseModelTypeVar], criteria: Type[CriteriaType] ) -> None: - """Validates that the types used in a criterion are compatible with the - types on the model. Raises a ValueError if they are not. + """Check if field_name is a field on a model. Raises a ValueError if it + is not. + + Raises + ------ + AttributeError + If the field_name is not a field on the model_type. + ValueError + If the field_value is not compatible with the field type. Notes ----- @@ -125,8 +122,8 @@ def _validate_criteria( """ if isinstance(criteria, Junction): for sub_criteria in criteria.members: - _validate_criteria(model_type, sub_criteria) - elif isinstance(criteria, Expression): + validate_criteria(model_type, sub_criteria) + else: # Expression if criteria.criterion["fieldName"] == "isNaFilter": _validate_field_name( model_type, @@ -168,8 +165,6 @@ def _validate_criteria( criteria.criterion["fieldName"], criteria.criterion["value"], ) - else: - raise ValueError(f"Invalid criterion type: {type(criteria)}") def resolve_filter_args( @@ -180,19 +175,16 @@ def resolve_filter_args( end: Optional[int] = None, **kwargs, ) -> tuple[list[Criterion], list[str], Optional[int], Optional[int]]: - """Validates filter arguments and resolves field names to SLIMS API + """Validates filter arguments and resolves field name aliases to SLIMS API column names. """ criteria = deepcopy(list(args)) criteria.extend(map(lambda item: equals(item[0], item[1]), kwargs.items())) resolved_criteria: list[Criterion] = [] for criterion in criteria: - _validate_criteria(model, criterion) - resolved_criteria.append(_resolve_criteria(model, criterion)) - resolved_sort = [ - resolve_model_alias(model, sort_key) for sort_key in sort - ] + validate_criteria(model, criterion) + resolved_criteria.append(resolve_criteria(model, criterion)) + resolved_sort = [resolve_model_alias(model, sort_key) for sort_key in sort] if start is not None and end is None or end is not None and start is None: - raise ValueError( - "Must provide both start and end or neither for fetch.") + raise ValueError("Must provide both start and end or neither for fetch.") return resolved_criteria, resolved_sort, start, end diff --git a/tests/test_core.py b/tests/test_core.py index 1c74b08..9b86975 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -8,16 +8,7 @@ from unittest.mock import MagicMock, patch from requests import Response -from slims.criteria import ( - between_inclusive, - conjunction, - disjunction, - equals, - is_na, - is_not, - is_not_one_of, - is_one_of, -) +from slims.criteria import conjunction, equals from slims.internal import Record, _SlimsApiException from aind_slims_api.core import SlimsAttachment, SlimsClient @@ -349,20 +340,6 @@ def test__validate_model_invalid_model(self, mock_log: MagicMock): self.assertEqual(1, len(validated)) self.assertEqual(1, mock_log.call_count) - def test_resolve_model_alias_invalid(self): - """Tests resolve_model_alias method raises expected error with an - invalid alias name. - """ - with self.assertRaises(ValueError): - self.example_client.resolve_model_alias(SlimsUnit, "not_an_alias") - - def test__validate_field_name_failure(self): - """Tests _validate_field_name method raises expected error with an - invalid field name. - """ - with self.assertRaises(ValueError): - self.example_client._validate_field_name(SlimsUnit, "not_an_alias") - @patch("slims.slims.Slims.fetch") def test_fetch_model_criterion(self, mock_slims_fetch: MagicMock): """Tests fetch_model method with a criterion.""" @@ -390,19 +367,6 @@ def test_fetch_model_criterion_invalid_criterion_type( self.example_client.fetch_model(SlimsUser, equals("username", 1)) mock_slims_fetch.assert_not_called() - @patch("slims.slims.Slims.fetch") - def test_fetch_model_invalid_criterion(self, mock_slims_fetch: MagicMock): - """Tests fetch_model method with a non Expression/Junction input.""" - mock_slims_fetch.return_value = [] - with self.assertRaises(ValueError): - self.example_client.fetch_model(SlimsUser, 1) - mock_slims_fetch.assert_not_called() - - def test__resolve_criteria_invalid_criterion(self): - """Tests _resolve_criteria method with a non Expression/Junction input.""" - with self.assertRaises(ValueError): - self.example_client._resolve_criteria(SlimsUser, 1) - @patch("slims.internal._SlimsApi.get_entities") def test_fetch_attachment(self, mock_get_entities: MagicMock): """Tests fetch_attachment method success.""" @@ -445,59 +409,6 @@ def test_fetch_models_invalid_start_end(self, mock_slims_fetch: MagicMock): self.example_client.fetch_models(SlimsUser._slims_table, end=1) mock_slims_fetch.assert_not_called() - def test_validate_criteria_is_one_of(self): - """Tests _validate_criteria method with an is_one_of, is_not_one_of - criterion. - """ - self.example_client._validate_criteria( - SlimsUser, - is_one_of("username", ["LKim", "JSmith"]), - ) - self.example_client._validate_criteria( - SlimsUser, - is_not_one_of("username", ["LKim", "JSmith"]), - ) - - def test_validate_criteria_between_inclusive(self): - """Tests _validate_criteria method with an between_inclusive criterion.""" - self.example_client._validate_criteria( - SlimsUser, - between_inclusive("username", "LKim", "JSmith"), - ) - - def test_validate_criteria_is_na(self): - """Tests _validate_criteria method with an is_na criterion.""" - self.example_client._validate_criteria( - SlimsUser, - is_na("username"), - ) - - def test_resolve_criteria_is_na(self): - """Tests _resolve_criteria method with an is_na criterion.""" - self.example_client._resolve_criteria( - SlimsUser, - is_na("username"), - ) - - def test_validate_criteria_is_not(self): - """Tests _validate_criteria method with an is_not criterion.""" - self.example_client._validate_criteria( - SlimsUser, - is_not(is_one_of("username", ["LKim", "JSmith"])), - ) - - def test_validate_criteria_disjunction(self): - """Tests _validate_criteria method with an is_not criterion.""" - criteria = ( - disjunction() - .add(is_one_of("username", ["LKim"])) - .add(is_one_of("username", ["JSmith"])) - ) - self.example_client._validate_criteria( - SlimsUser, - criteria, - ) - if __name__ == "__main__": unittest.main() diff --git a/tests/test_filters.py b/tests/test_filters.py new file mode 100644 index 0000000..338f9cd --- /dev/null +++ b/tests/test_filters.py @@ -0,0 +1,94 @@ +"""Tests functions in filters""" + +import unittest + +from slims.criteria import ( + between_inclusive, + disjunction, + is_na, + is_not, + is_not_one_of, + is_one_of, +) + +from aind_slims_api import filters +from aind_slims_api.models.unit import SlimsUnit +from aind_slims_api.models.user import SlimsUser + + +class TestFilters(unittest.TestCase): + """Tests filter validation and utility functions.""" + + def test_resolve_model_alias_invalid(self): + """Tests resolve_model_alias method raises expected error with an + invalid alias name. + """ + with self.assertRaises(ValueError): + filters.resolve_model_alias(SlimsUnit, "not_an_alias") + + def test__validate_field_name_failure(self): + """Tests _validate_field_name method raises expected error with an + invalid field name. + """ + with self.assertRaises(ValueError): + filters._validate_field_name(SlimsUnit, "not_an_alias") + + def test_validate_criteria_is_one_of(self): + """Tests _validate_criteria method with an is_one_of criterion.""" + filters.validate_criteria( + SlimsUser, + is_one_of("username", ["LKim", "JSmith"]), + ) + + def test_validate_criteria_is_not_one_of(self): + """Tests _validate_criteria method with an is_not_one_of + criterion. + """ + filters.validate_criteria( + SlimsUser, + is_not_one_of("username", ["LKim", "JSmith"]), + ) + + def test_validate_criteria_between_inclusive(self): + """Tests _validate_criteria method with an between_inclusive criterion.""" + filters.validate_criteria( + SlimsUser, + between_inclusive("username", "LKim", "JSmith"), + ) + + def test_validate_criteria_is_na(self): + """Tests _validate_criteria method with an is_na criterion.""" + filters.validate_criteria( + SlimsUser, + is_na("username"), + ) + + def test_resolve_criteria_is_na(self): + """Tests _resolve_criteria method with an is_na criterion.""" + filters.resolve_criteria( + SlimsUser, + is_na("username"), + ) + + def test_validate_criteria_is_not(self): + """Tests _validate_criteria method with an is_not criterion.""" + filters.validate_criteria( + SlimsUser, + is_not(is_one_of("username", ["LKim", "JSmith"])), + ) + + def test_validate_criteria_disjunction(self): + """Tests _validate_criteria method with an is_not criterion.""" + criteria = ( + disjunction() + .add(is_one_of("username", ["LKim"])) + .add(is_one_of("username", ["JSmith"])) + ) + filters.validate_criteria( + SlimsUser, + criteria, + ) + + +if __name__ == "__main__": + unittest.main()