diff --git a/src/aind_slims_api/core.py b/src/aind_slims_api/core.py index c76644a..63e422f 100644 --- a/src/aind_slims_api/core.py +++ b/src/aind_slims_api/core.py @@ -12,7 +12,7 @@ import logging from copy import deepcopy from functools import lru_cache -from typing import Optional, Type, TypeVar, get_type_hints +from typing import Any, Optional, Type, TypeVar, get_type_hints from pydantic import ValidationError from requests import Response @@ -167,6 +167,35 @@ def _resolve_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 @@ -178,23 +207,48 @@ def _validate_criteria( for sub_criteria in criteria.members: SlimsClient._validate_criteria(model_type, sub_criteria) elif isinstance(criteria, Expression): - field_type_map = get_type_hints(model_type) - field_name = criteria.criterion["fieldName"] - value = criteria.criterion["value"] - field_type = field_type_map[field_name] - if not isinstance(value, field_type): - raise ValueError( - f"{value} is incompatible with {field_type}" - f" for field {field_name}" - ) + 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], *args: Criterion, - sort: Optional[str | list[str]] = None, + sort: str | list[str] = [], start: Optional[int] = None, end: Optional[int] = None, **kwargs, @@ -212,31 +266,23 @@ def fetch_models( - kwargs are mapped to field alias names and used as equality filters for the fetch. """ - resolved_kwargs = deepcopy(model._base_fetch_filters) - for name, value in kwargs.items(): - resolved_kwargs[self.resolve_model_alias(model, name)] = value - logger.debug("Resolved kwargs: %s", resolved_kwargs) - resolved_sort: Optional[str | list[str]] = None - if sort is not None: - if isinstance(sort, str): - resolved_sort = self.resolve_model_alias(model, sort) - else: - resolved_sort = [ - self.resolve_model_alias(model, sort_key) for sort_key in sort - ] - logger.debug("Resolved sort: %s", resolved_sort) - resolved_args = [] - for arg in args: - self._validate_criteria(model, arg) - resolved_args.append(self._resolve_criteria(model, arg)) - logger.debug("Resolved args: %s", resolved_args) + if isinstance(sort, str): + sort = [sort] + + criteria, resolved_sort, start, end = self._resolve_filter_args( + model, + *args, + sort=sort, + start=start, + end=end, + **kwargs, + ) response = self.fetch( model._slims_table, - *resolved_args, + *criteria, sort=resolved_sort, start=start, end=end, - **resolved_kwargs, ) return self._validate_models(model, response) @@ -252,6 +298,7 @@ def fetch_model( Notes ----- - kwargs are mapped to field alias values + - sorts records on created_on in descending order and returns the first """ records = self.fetch_models( model, @@ -267,18 +314,93 @@ def fetch_model( raise SlimsRecordNotFound("No record found.") return records[0] + @staticmethod + def _create_get_entities_body( + *args: Criterion, + sort: list[str] = [], + start: Optional[int] = None, + end: Optional[int] = None, + ) -> dict[str, Any]: + """Creates get entities body for SLIMS API request.""" + body: dict[str, Any] = { + "sortBy": sort, + "startRow": start, + "endRow": end, + } + if args: + criteria = conjunction() + for arg in args: + criteria.add(arg) + body["criteria"] = criteria.to_dict() + + return body + def fetch_attachments( self, record: SlimsBaseModel, + *args, + sort: str | list[str] = [], + start: Optional[int] = None, + end: Optional[int] = None, + **kwargs, ) -> list[SlimsAttachment]: - """Fetch attachments for a given record.""" + """Fetch attachments for a given record. + + Notes + ----- + - kwargs are mapped to field alias values + """ + if isinstance(sort, str): + sort = [sort] + + criteria, sort, start, end = self._resolve_filter_args( + SlimsAttachment, + *args, + sort=sort, + start=start, + end=end, + **kwargs, + ) return self._validate_models( SlimsAttachment, self.db.slims_api.get_entities( - f"attachment/{record._slims_table}/{record.pk}" + f"attachment/{record._slims_table}/{record.pk}", + body=self._create_get_entities_body( + *criteria, + sort=sort, + start=start, + end=end, + ), ), ) + def fetch_attachment( + self, + record: SlimsBaseModel, + *args, + **kwargs, + ) -> SlimsAttachment: + """Fetch attachments for a given record. + + Notes + ----- + - kwargs are mapped to field alias values + - sorts records on created_on in descending order and returns the first + """ + records = self.fetch_attachments( + record, + *args, + sort="-created_on", + start=0, # slims rows appear to be 0-indexed + end=1, + **kwargs, + ) + if len(records) > 0: + logger.debug(f"Found {len(records)} records for {record}.") + if len(records) < 1: + raise SlimsRecordNotFound("No record found.") + return records[0] + def fetch_attachment_content( self, attachment: int | SlimsAttachment, diff --git a/src/aind_slims_api/models/metadata.py b/src/aind_slims_api/models/metadata.py index 6a44617..35a2a9e 100644 --- a/src/aind_slims_api/models/metadata.py +++ b/src/aind_slims_api/models/metadata.py @@ -28,6 +28,11 @@ class SlimsMetadataReference(SlimsBaseModel): >>> metadata.json()["rig_id"] '323_EPHYS1_OPTO_2024-02-12' + ### Read latest attachment + >>> latest_attachment = client.fetch_attachment( + ... metadata_reference, + ... ) + ### Write >>> import json >>> attachment_pk = client.add_attachment_content( diff --git a/tests/test_core.py b/tests/test_core.py index b101dd3..30150f8 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -347,6 +347,13 @@ def test_resolve_model_alias_invalid(self): 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.""" @@ -387,6 +394,48 @@ def test__resolve_criteria_invalid_criterion(self): 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.""" + mock_get_entities.return_value = self.example_fetch_attachment_response + unit = SlimsUnit.model_validate( + Record( + json_entity=self.example_fetch_unit_response[0].json_entity, + slims_api=self.example_client.db.slims_api, + ) + ) + self.example_client.fetch_attachment(unit, equals("name", "test")) + + @patch("slims.internal._SlimsApi.get_entities") + def test_fetch_attachment_no_attachments(self, mock_get_entities: MagicMock): + """Tests fetch_attachment method failure due to no attachments.""" + mock_get_entities.return_value = [] + unit = SlimsUnit.model_validate( + Record( + json_entity=self.example_fetch_unit_response[0].json_entity, + slims_api=self.example_client.db.slims_api, + ) + ) + with self.assertRaises(SlimsRecordNotFound): + self.example_client.fetch_attachment(unit) + + @patch("slims.slims.Slims.fetch") + def test_fetch_str_sort(self, mock_slims_fetch: MagicMock): + """Tests fetch method when sort is a string.""" + mock_slims_fetch.return_value = [] + self.example_client.fetch(SlimsUser._slims_table, sort="username") + mock_slims_fetch.assert_called_once() + + @patch("slims.slims.Slims.fetch") + def test_fetch_models_invalid_start_end(self, mock_slims_fetch: MagicMock): + """Tests fetch_model method failure due to only supplying start or end.""" + mock_slims_fetch.return_value = [] + with self.assertRaises(ValueError): + self.example_client.fetch_models(SlimsUser._slims_table, start=1) + with self.assertRaises(ValueError): + self.example_client.fetch_models(SlimsUser._slims_table, end=1) + mock_slims_fetch.assert_not_called() + if __name__ == "__main__": unittest.main()