diff --git a/README.md b/README.md index e1a8d80..58a7074 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,13 @@ To develop the code, run pip install -e .[dev] ``` +## Usage + +### VScode +![](example-usage.gif) + +Requires the [pylance extension](https://marketplace.visualstudio.com/items?itemName=ms-python.vscode-pylance) to be installed for similar functionality. + ## Contributing ### Linters and testing diff --git a/example-usage.gif b/example-usage.gif new file mode 100644 index 0000000..a1efa75 Binary files /dev/null and b/example-usage.gif differ diff --git a/src/aind_slims_api/behavior_session.py b/src/aind_slims_api/behavior_session.py deleted file mode 100644 index cfaff44..0000000 --- a/src/aind_slims_api/behavior_session.py +++ /dev/null @@ -1,123 +0,0 @@ -"""Contains a model for the behavior session content events, a method for - fetching it and writing it. -""" - -import logging -from typing import Any -from datetime import datetime - -from pydantic import Field - -from aind_slims_api.core import SlimsBaseModel, SlimsClient, SLIMSTABLES - -logger = logging.getLogger() - - -class SlimsBehaviorSessionContentEvent(SlimsBaseModel): - """Model for an instance of the Behavior Session ContentEvent""" - - pk: int | None = Field(default=None, alias="cnvn_pk") - mouse_pk: int | None = Field( - default=None, alias="cnvn_fk_content" - ) # used as reference to mouse - notes: str | None = Field(default=None, alias="cnvn_cf_notes") - task_stage: str | None = Field(default=None, alias="cnvn_cf_taskStage") - instrument: int | None = Field(default=None, alias="cnvn_cf_fk_instrument") - trainers: list[int] = Field(default=[], alias="cnvn_cf_fk_trainer") - task: str | None = Field(default=None, alias="cnvn_cf_task") - is_curriculum_suggestion: bool | None = Field( - default=None, alias="cnvn_cf_stageIsOnCurriculum" - ) - task_schema_version: str | None = Field( - default=None, alias="cnvn_cf_taskSchemaVersion" - ) - software_version: str | None = Field(default=None, alias="cnvn_cf_softwareVersion") - date: datetime | None = Field(..., alias="cnvn_cf_scheduledDate") - - cnvn_fk_contentEventType: int = 10 # pk of Behavior Session ContentEvent - - _slims_table: SLIMSTABLES = "ContentEvent" - - -SlimsSingletonFetchReturn = SlimsBaseModel | dict[str, Any] | None - - -def _resolve_pk( - model: SlimsSingletonFetchReturn, - primary_key_name: str = "pk", -) -> int: - """Utility function shared across read/write - - Notes - ----- - - TODO: Change return type of fetch_mouse_content to match pattern in - fetch_behavior_session_content_events, or the other way around? - - TODO: Move to core to have better centralized control of when references - are resolved - """ - if isinstance(model, dict): - logger.warning("Extracting primary key from unvalidated dict.") - return model[primary_key_name] - elif isinstance(model, SlimsBaseModel): - return getattr(model, primary_key_name) - elif model is None: - raise ValueError(f"Cannot resolve primary key from {model}") - else: - raise ValueError("Unexpected type for model: %s" % type(model)) - - -def fetch_behavior_session_content_events( - client: SlimsClient, - mouse: SlimsSingletonFetchReturn, -) -> tuple[list[SlimsBehaviorSessionContentEvent], list[dict[str, Any]]]: - """Fetches behavior sessions for a mouse with labtracks id {mouse_name} - - Returns - ------- - tuple: - list: - Validated SlimsBehaviorSessionContentEvent objects - list: - Dictionaries representations of objects that failed validation - """ - return client.fetch_models( - SlimsBehaviorSessionContentEvent, - cnvn_fk_content=_resolve_pk(mouse), - cnvt_name="Behavior Session", - sort=["cnvn_cf_scheduledDate"], - ) - - -def write_behavior_session_content_events( - client: SlimsClient, - mouse: SlimsSingletonFetchReturn, - instrument: SlimsSingletonFetchReturn, - trainers: list[SlimsSingletonFetchReturn], - *behavior_sessions: SlimsBehaviorSessionContentEvent, -) -> list[SlimsBehaviorSessionContentEvent]: - """Writes behavior sessions for a mouse with labtracks id {mouse_name} - - Notes - ----- - - All supplied `behavior_sessions` will have their `mouse_name` field set - to the value supplied as `mouse_name` to this function - """ - mouse_pk = _resolve_pk(mouse) - logger.debug(f"Mouse pk: {mouse_pk}") - instrument_pk = _resolve_pk(instrument) - logger.debug(f"Instrument pk: {instrument_pk}") - trainer_pks = [_resolve_pk(trainer) for trainer in trainers] - logger.debug(f"Trainer pks: {trainer_pks}") - added = [] - for behavior_session in behavior_sessions: - updated = behavior_session.model_copy( - update={ - "mouse_pk": mouse_pk, - "instrument": instrument_pk, - "trainers": trainer_pks, - }, - ) - logger.debug(f"Resolved behavior session: {updated}") - added.append(client.add_model(updated)) - - return added diff --git a/src/aind_slims_api/configuration.py b/src/aind_slims_api/configuration.py index 6b6d62e..fd5070b 100644 --- a/src/aind_slims_api/configuration.py +++ b/src/aind_slims_api/configuration.py @@ -13,4 +13,4 @@ class AindSlimsApiSettings(BaseSettings): slims_url: str = "https://aind-test.us.slims.agilent.com/slimsrest/" slims_username: str = "" - slims_password: SecretStr = "" + slims_password: SecretStr = SecretStr("") diff --git a/src/aind_slims_api/core.py b/src/aind_slims_api/core.py index b94844e..b8932d9 100644 --- a/src/aind_slims_api/core.py +++ b/src/aind_slims_api/core.py @@ -8,130 +8,24 @@ methods and integration with SlimsBaseModel subtypes """ -from datetime import datetime -from functools import lru_cache -from pydantic import ( - BaseModel, - ValidationInfo, - ValidationError, - field_serializer, - field_validator, -) -from pydantic.fields import FieldInfo import logging -from typing import Any, Literal, Optional, Type, TypeVar +from copy import deepcopy +from functools import lru_cache +from typing import Optional, Type, TypeVar -from slims.slims import Slims, _SlimsApiException -from slims.internal import ( - Column as SlimsColumn, - Record as SlimsRecord, -) +from pydantic import ValidationError +from requests import Response 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.models.attachment import SlimsAttachment +from aind_slims_api.models.base import SlimsBaseModel +from aind_slims_api.types import SLIMS_TABLES -logger = logging.getLogger() - -# List of slims tables manually accessed, there are many more -SLIMSTABLES = Literal[ - "Project", - "Content", - "ContentEvent", - "Unit", - "Result", - "Test", - "User", - "Groups", - "Instrument", -] - - -class UnitSpec: - """Used in type annotation metadata to specify units""" - - units: list[str] - preferred_unit: str = None - - def __init__(self, *args, preferred_unit=None): - """Set list of acceptable units from args, and preferred_unit""" - self.units = args - if len(self.units) == 0: - raise ValueError("One or more units must be specified") - if preferred_unit is None: - self.preferred_unit = self.units[0] - - -def _find_unit_spec(field: FieldInfo) -> UnitSpec | None: - """Given a Pydantic FieldInfo, find the UnitSpec in its metadata""" - metadata = field.metadata - for m in metadata: - if isinstance(m, UnitSpec): - return m - return None - - -class SlimsBaseModel( - BaseModel, - from_attributes=True, - validate_assignment=True, -): - """Pydantic model to represent a SLIMS record. - Subclass with fields matching those in the SLIMS record. - - For Quantities, specify acceptable units like so: - - class MyModel(SlimsBaseModel): - myfield: Annotated[float | None, UnitSpec("g","kg")] - - Quantities will be serialized using the first unit passed - - Datetime fields will be serialized to an integer ms timestamp - """ - - pk: int = None - json_entity: dict = None - _slims_table: SLIMSTABLES - - @field_validator("*", mode="before") - def _validate(cls, value, info: ValidationInfo): - """Validates a field, accounts for Quantities""" - if isinstance(value, SlimsColumn): - if value.datatype == "QUANTITY": - unit_spec = _find_unit_spec(cls.model_fields[info.field_name]) - if unit_spec is None: - msg = ( - f'Quantity field "{info.field_name}"' - "must be annotated with a UnitSpec" - ) - raise TypeError(msg) - if value.unit not in unit_spec.units: - msg = ( - f'Unexpected unit "{value.unit}" for field ' - f"{info.field_name}, Expected {unit_spec.units}" - ) - raise ValueError(msg) - return value.value - else: - return value - - @field_serializer("*") - def _serialize(self, field, info): - """Serialize a field, accounts for Quantities and datetime""" - unit_spec = _find_unit_spec(self.model_fields[info.field_name]) - if unit_spec and field is not None: - quantity = { - "amount": field, - "unit_display": unit_spec.preferred_unit, - } - return quantity - elif isinstance(field, datetime): - return int(field.timestamp() * 10**3) - else: - return field - - # TODO: Add links - need Record.json_entity['links']['self'] - # TODO: Add Table - need Record.json_entity['tableName'] - # TODO: Support attachments +logger = logging.getLogger(__name__) SlimsBaseModelTypeVar = TypeVar("SlimsBaseModelTypeVar", bound=SlimsBaseModel) @@ -140,10 +34,11 @@ def _serialize(self, field, info): class SlimsClient: """Wrapper around slims-python-api client with convenience methods""" + db: Slims + def __init__(self, url=None, username=None, password=None): """Create object and try to connect to database""" self.url = url or config.slims_url - self.db: Optional[Slims] = None self.connect( self.url, @@ -162,7 +57,7 @@ def connect(self, url: str, username: str, password: str): def fetch( self, - table: SLIMSTABLES, + table: SLIMS_TABLES, *args, sort: Optional[str | list[str]] = None, start: Optional[int] = None, @@ -204,6 +99,32 @@ 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""" + for field_name, field_info in model.model_fields.items(): + if field_name == attr_name and field_info.alias: + return field_info.alias + else: + raise ValueError(f"Cannot resolve alias for {attr_name} on {model}") + + @staticmethod + def _validate_models( + model_type: Type[SlimsBaseModelTypeVar], records: list[SlimsRecord] + ) -> list[SlimsBaseModelTypeVar]: + """Validate a list of SlimsBaseModel objects. Logs errors for records + that fail pydantic validation.""" + validated = [] + for record in records: + try: + validated.append(model_type.model_validate(record)) + except ValidationError as e: + logger.error(f"SLIMS data validation failed, {repr(e)}") + return validated + def fetch_models( self, model: Type[SlimsBaseModelTypeVar], @@ -212,7 +133,7 @@ def fetch_models( start: Optional[int] = None, end: Optional[int] = None, **kwargs, - ) -> tuple[list[SlimsBaseModelTypeVar], list[dict[str, Any]]]: + ) -> list[SlimsBaseModelTypeVar]: """Fetch records from SLIMS and return them as SlimsBaseModel objects Returns @@ -220,30 +141,82 @@ def fetch_models( tuple: list: Validated SlimsBaseModel objects - list: - Dictionaries representations of objects that failed validation + + Notes + ----- + - kwargs are mapped to field alias values """ + 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) response = self.fetch( - model._slims_table.default, # TODO: consider changing fetch method + model._slims_table, # TODO: consider changing fetch method + *args, + sort=resolved_sort, + start=start, + end=end, + **resolved_kwargs, + ) + return self._validate_models(model, response) + + def fetch_model( + self, + model: Type[SlimsBaseModelTypeVar], + *args, + sort: Optional[str | list[str]] = None, + start: Optional[int] = None, + end: Optional[int] = None, + **kwargs, + ) -> SlimsBaseModelTypeVar | None: + """Fetch a single record from SLIMS and return it as a validated + SlimsBaseModel object. + + Notes + ----- + - kwargs are mapped to field alias values + """ + records = self.fetch_models( + model, *args, sort=sort, start=start, end=end, **kwargs, ) - validated = [] - unvalidated = [] - for record in response: - try: - validated.append(model.model_validate(record)) - except ValidationError as e: - logger.error(f"SLIMS data validation failed, {repr(e)}") - unvalidated.append(record.json_entity) + if len(records) > 0: + logger.debug(f"Found {len(records)} records for {model}.") + if len(records) < 1: + raise SlimsRecordNotFound("No record found.") + return records[0] - return validated, unvalidated + def fetch_attachments( + self, + record: SlimsBaseModel, + ) -> list[SlimsAttachment]: + """Fetch attachments for a given record.""" + return self._validate_models( + SlimsAttachment, + self.db.slims_api.get_entities( + f"attachment/{record._slims_table}/{record.pk}" + ), + ) + + def fetch_attachment_content(self, attachment: SlimsAttachment) -> Response: + """Fetch attachment content for a given attachment.""" + return self.db.slims_api.get(f"repo/{attachment.pk}") @lru_cache(maxsize=None) - def fetch_pk(self, table: SLIMSTABLES, *args, **kwargs) -> int | None: + def fetch_pk(self, table: SLIMS_TABLES, *args, **kwargs) -> int | None: """SlimsClient.fetch but returns the pk of the first returned record""" records = self.fetch(table, *args, **kwargs) if len(records) > 0: @@ -255,13 +228,13 @@ def fetch_user(self, user_name: str): """Fetches a user by username""" return self.fetch("User", user_userName=user_name) - def add(self, table: SLIMSTABLES, data: dict): + def add(self, table: SLIMS_TABLES, data: dict): """Add a SLIMS record to a given SLIMS table""" record = self.db.add(table, data) logger.info(f"SLIMS Add: {table}/{record.pk()}") return record - def update(self, table: SLIMSTABLES, pk: int, data: dict): + def update(self, table: SLIMS_TABLES, pk: int, data: dict): """Update a SLIMS record""" record = self.db.fetch_by_pk(table, pk) if record is None: @@ -270,7 +243,7 @@ def update(self, table: SLIMSTABLES, pk: int, data: dict): logger.info(f"SLIMS Update: {table}/{pk}") return new_record - def rest_link(self, table: SLIMSTABLES, **kwargs): + def rest_link(self, table: SLIMS_TABLES, **kwargs): """Construct a url link to a SLIMS table with arbitrary filters""" base_url = f"{self.url}/rest/{table}" queries = [f"?{k}={v}" for k, v in kwargs.items()] @@ -291,7 +264,7 @@ def add_model( """ fields_to_include = set(args) or None fields_to_exclude = set(kwargs.get("exclude", [])) - fields_to_exclude.add("pk") + fields_to_exclude.update(["pk", "attachments", "slims_api"]) rtn = self.add( model._slims_table, model.model_dump( @@ -315,6 +288,9 @@ def update_model(self, model: SlimsBaseModel, *args, **kwargs): An instance of the same type of model, with data from the resulting SLIMS record """ + if model.pk is None: + raise ValueError("Cannot update model without a pk") + fields_to_include = set(args) or None rtn = self.update( model._slims_table, diff --git a/src/aind_slims_api/exceptions.py b/src/aind_slims_api/exceptions.py new file mode 100644 index 0000000..0442748 --- /dev/null +++ b/src/aind_slims_api/exceptions.py @@ -0,0 +1,9 @@ +"""Custom exceptions for the AIND Slims API.""" + + +class SlimsAPIException(Exception): + """Base exception for the AIND Slims API.""" + + +class SlimsRecordNotFound(SlimsAPIException): + """Exception raised when a record is not found in the SLIMS database.""" diff --git a/src/aind_slims_api/instrument.py b/src/aind_slims_api/instrument.py deleted file mode 100644 index 4314fcf..0000000 --- a/src/aind_slims_api/instrument.py +++ /dev/null @@ -1,66 +0,0 @@ -"""Contains a model for the instrument content, and a method for fetching it""" - -import logging -from typing import Any - -from pydantic import Field - -from aind_slims_api.core import SlimsBaseModel, SlimsClient, SLIMSTABLES - -logger = logging.getLogger() - - -class SlimsInstrument(SlimsBaseModel): - """Model for an instance of the Behavior Session ContentEvent""" - - name: str = Field(..., alias="nstr_name") - pk: int = Field(..., alias="nstr_pk") - _slims_table: SLIMSTABLES = "Instrument" - - # todo add more useful fields - - -def fetch_instrument_content( - client: SlimsClient, - instrument_name: str, -) -> SlimsInstrument | dict[str, Any] | None: - """Fetches behavior sessions for a mouse with labtracks id {mouse_name} - - Returns - ------- - tuple: - list: - Validated SlimsInstrument objects - list: - Dictionaries representations of objects that failed validation - - Notes - ----- - - Todo: add partial name match or some other type of filtering - - TODO: reconsider this pattern, consider just returning all records or - having number returned be a parameter or setting - """ - validated, unvalidated = client.fetch_models( - SlimsInstrument, - nstr_name=instrument_name, - ) - if len(validated) > 0: - validated_details = validated[0] - if len(validated) > 1: - logger.warning( - f"Warning, Multiple instruments in SLIMS with name {instrument_name}, " - f"using pk={validated_details.pk}" - ) - return validated_details - else: - if len(unvalidated) > 0: - unvalidated_details = unvalidated[0] - if len(unvalidated) > 1: - logger.warning( - "Warning, Multiple instruments in SLIMS with name " - f"{instrument_name}, " - f"using pk={unvalidated_details['pk']}" - ) - return unvalidated[0] - - return None diff --git a/src/aind_slims_api/models/__init__.py b/src/aind_slims_api/models/__init__.py new file mode 100644 index 0000000..5a85933 --- /dev/null +++ b/src/aind_slims_api/models/__init__.py @@ -0,0 +1,19 @@ +"""Models for abstractions around Slims records. Friendlier names and + documenation on how things relate to each other in our specific instance. +""" + +from aind_slims_api.models.attachment import SlimsAttachment +from aind_slims_api.models.behavior_session import SlimsBehaviorSession +from aind_slims_api.models.instrument import SlimsInstrument +from aind_slims_api.models.mouse import SlimsMouseContent +from aind_slims_api.models.unit import SlimsUnit +from aind_slims_api.models.user import SlimsUser + +__all__ = [ + "SlimsAttachment", + "SlimsBehaviorSession", + "SlimsInstrument", + "SlimsMouseContent", + "SlimsUnit", + "SlimsUser", +] diff --git a/src/aind_slims_api/models/attachment.py b/src/aind_slims_api/models/attachment.py new file mode 100644 index 0000000..8577d27 --- /dev/null +++ b/src/aind_slims_api/models/attachment.py @@ -0,0 +1,28 @@ +"""Model for a record in the Attachment table in SLIMS.""" + +from pydantic import Field + +from aind_slims_api.models.base import SlimsBaseModel + + +class SlimsAttachment(SlimsBaseModel): + """Model for a record in the Attachment table in SLIMS. + + Examples + -------- + >>> from aind_slims_api import SlimsClient + >>> client = SlimsClient() + >>> rig_metadata_attachment = client.fetch_model( + ... SlimsAttachment, + ... name="rig323_EPHYS1_OPTO_2024-02-12.json" + ... ) + >>> rig_metadata = client.fetch_attachment_content( + ... rig_metadata_attachment + ... ).json() + >>> rig_metadata["rig_id"] + '323_EPHYS1_OPTO_2024-02-12' + """ + + pk: int = Field(..., alias="attm_pk") + name: str = Field(..., alias="attm_name") + _slims_table = "Attachment" diff --git a/src/aind_slims_api/models/base.py b/src/aind_slims_api/models/base.py new file mode 100644 index 0000000..c011633 --- /dev/null +++ b/src/aind_slims_api/models/base.py @@ -0,0 +1,79 @@ +"""Base model for SLIMS records abstraction. +""" + +import logging +from datetime import datetime +from typing import ClassVar, Optional + +from pydantic import BaseModel, ValidationInfo, field_serializer, field_validator +from slims.internal import Column as SlimsColumn + +from aind_slims_api.models.utils import _find_unit_spec +from aind_slims_api.types import SLIMS_TABLES + +logger = logging.getLogger(__name__) + + +class SlimsBaseModel( + BaseModel, + from_attributes=True, + validate_assignment=True, +): + """Pydantic model to represent a SLIMS record. + Subclass with fields matching those in the SLIMS record. + + For Quantities, specify acceptable units like so: + + class MyModel(SlimsBaseModel): + myfield: Annotated[float | None, UnitSpec("g","kg")] + + Quantities will be serialized using the first unit passed + + Datetime fields will be serialized to an integer ms timestamp + """ + + pk: Optional[int] = None + json_entity: Optional[dict] = None + _slims_table: ClassVar[SLIMS_TABLES] + # base filters for model fetch + _base_fetch_filters: ClassVar[dict[str, str]] = {} + + @field_validator("*", mode="before") + def _validate(cls, value, info: ValidationInfo): + """Validates a field, accounts for Quantities""" + if isinstance(value, SlimsColumn): + if value.datatype == "QUANTITY": + unit_spec = _find_unit_spec(cls.model_fields[info.field_name]) + if unit_spec is None: + msg = ( + f'Quantity field "{info.field_name}"' + "must be annotated with a UnitSpec" + ) + raise TypeError(msg) + if value.unit not in unit_spec.units: + msg = ( + f'Unexpected unit "{value.unit}" for field ' + f"{info.field_name}, Expected {unit_spec.units}" + ) + raise ValueError(msg) + return value.value + else: + return value + + @field_serializer("*") + def _serialize(self, field, info): + """Serialize a field, accounts for Quantities and datetime""" + unit_spec = _find_unit_spec(self.model_fields[info.field_name]) + if unit_spec and field is not None: + quantity = { + "amount": field, + "unit_display": unit_spec.preferred_unit, + } + return quantity + elif isinstance(field, datetime): + return int(field.timestamp() * 10**3) + else: + return field + + # TODO: Add links - need Record.json_entity['links']['self'] + # TODO: Add Table - need Record.json_entity['tableName'] diff --git a/src/aind_slims_api/models/behavior_session.py b/src/aind_slims_api/models/behavior_session.py new file mode 100644 index 0000000..fe6ec67 --- /dev/null +++ b/src/aind_slims_api/models/behavior_session.py @@ -0,0 +1,54 @@ +"""Contains a model for the behavior session content events, a method for + fetching it and writing it. +""" + +import logging +from datetime import datetime +from typing import ClassVar + +from pydantic import Field + +from aind_slims_api.models.base import SlimsBaseModel + +logger = logging.getLogger() + + +class SlimsBehaviorSession(SlimsBaseModel): + """Model for an instance of the Behavior Session ContentEvent + + Examples + -------- + >>> from aind_slims_api import SlimsClient + >>> from aind_slims_api.models import SlimsMouseContent + >>> client = SlimsClient() + >>> mouse = client.fetch_model(SlimsMouseContent, barcode="00000000") + >>> behavior_sessions = client.fetch_models(SlimsBehaviorSession, + ... mouse_pk=mouse.pk, sort=["date"]) + """ + + pk: int | None = Field(default=None, alias="cnvn_pk") + mouse_pk: int | None = Field( + default=None, + alias="cnvn_fk_content", + description=( + "The primary key of the mouse associated with this behavior session." + ), + ) # used as reference to mouse + notes: str | None = Field(default=None, alias="cnvn_cf_notes") + task_stage: str | None = Field(default=None, alias="cnvn_cf_taskStage") + instrument: int | None = Field(default=None, alias="cnvn_cf_fk_instrument") + trainers: list[int] = Field(default=[], alias="cnvn_cf_fk_trainer") + task: str | None = Field(default=None, alias="cnvn_cf_task") + is_curriculum_suggestion: bool | None = Field( + default=None, alias="cnvn_cf_stageIsOnCurriculum" + ) + task_schema_version: str | None = Field( + default=None, alias="cnvn_cf_taskSchemaVersion" + ) + software_version: str | None = Field(default=None, alias="cnvn_cf_softwareVersion") + date: datetime | None = Field(default=None, alias="cnvn_cf_scheduledDate") + cnvn_fk_contentEventType: int = 10 # pk of Behavior Session ContentEvent + _slims_table = "ContentEvent" + _base_fetch_filters: ClassVar[dict[str, str]] = { + "cnvt_name": "Behavior Session", + } diff --git a/src/aind_slims_api/models/instrument.py b/src/aind_slims_api/models/instrument.py new file mode 100644 index 0000000..bb47ce4 --- /dev/null +++ b/src/aind_slims_api/models/instrument.py @@ -0,0 +1,26 @@ +"""Contains a model for the instrument content, and a method for fetching it""" + +from pydantic import Field + +from aind_slims_api.models.base import SlimsBaseModel + + +class SlimsInstrument(SlimsBaseModel): + """Model for a SLIMS instrument record. + + Examples + -------- + >>> from aind_slims_api.core import SlimsClient + >>> client = SlimsClient() + >>> instrument = client.fetch_model(SlimsInstrument, name="323_EPHYS1_OPTO") + """ + + name: str = Field( + ..., + alias="nstr_name", + description="The name of the instrument", + ) + pk: int = Field(..., alias="nstr_pk") + _slims_table = "Instrument" + + # todo add more useful fields diff --git a/src/aind_slims_api/models/mouse.py b/src/aind_slims_api/models/mouse.py new file mode 100644 index 0000000..33a5442 --- /dev/null +++ b/src/aind_slims_api/models/mouse.py @@ -0,0 +1,54 @@ +"""Contains a model for the mouse content, and a method for fetching it""" + +from typing import Annotated, ClassVar + +from pydantic import BeforeValidator, Field + +from aind_slims_api.models.base import SlimsBaseModel +from aind_slims_api.models.utils import UnitSpec + + +class SlimsMouseContent(SlimsBaseModel): + """Model for an instance of the Mouse ContentType + + Properties + ---------- + barcode: str, barcode of the mouse, filterable + + Examples + -------- + >>> from aind_slims_api.core import SlimsClient + >>> client = SlimsClient() + >>> mouse = client.fetch_model(SlimsMouseContent, barcode="00000000") + """ + + baseline_weight_g: Annotated[float | None, UnitSpec("g")] = Field( + ..., alias="cntn_cf_baselineWeight" + ) + point_of_contact: str | None = Field(..., alias="cntn_cf_scientificPointOfContact") + water_restricted: Annotated[bool, BeforeValidator(lambda x: x or False)] = Field( + ..., alias="cntn_cf_waterRestricted" + ) + barcode: str = Field(..., alias="cntn_barCode") + pk: int = Field(..., alias="cntn_pk") + + _slims_table = "Content" + _base_fetch_filters: ClassVar[dict[str, str]] = { + "cntp_name": "Mouse", + } + + # TODO: Include other helpful fields (genotype, gender...) + + # pk: callable + # cntn_fk_category: SlimsColumn + # cntn_fk_contentType: SlimsColumn + # cntn_barCode: SlimsColumn + # cntn_id: SlimsColumn + # cntn_cf_contactPerson: SlimsColumn + # cntn_status: SlimsColumn + # cntn_fk_status: SlimsColumn + # cntn_fk_user: SlimsColumn + # cntn_cf_fk_fundingCode: SlimsColumn + # cntn_cf_genotype: SlimsColumn + # cntn_cf_labtracksId: SlimsColumn + # cntn_cf_parentBarcode: SlimsColumn diff --git a/src/aind_slims_api/unit.py b/src/aind_slims_api/models/unit.py similarity index 73% rename from src/aind_slims_api/unit.py rename to src/aind_slims_api/models/unit.py index 73c2efc..e3771bc 100644 --- a/src/aind_slims_api/unit.py +++ b/src/aind_slims_api/models/unit.py @@ -1,13 +1,10 @@ """Contains a model for a unit""" -import logging from typing import Optional from pydantic import Field -from aind_slims_api.core import SlimsBaseModel - -logger = logging.getLogger() +from aind_slims_api.models.base import SlimsBaseModel class SlimsUnit(SlimsBaseModel): @@ -17,4 +14,4 @@ class SlimsUnit(SlimsBaseModel): abbreviation: Optional[str] = Field("", alias="unit_abbreviation") pk: int = Field(..., alias="unit_pk") - _slims_table: str = "Unit" + _slims_table = "Unit" diff --git a/src/aind_slims_api/models/user.py b/src/aind_slims_api/models/user.py new file mode 100644 index 0000000..e8eb09b --- /dev/null +++ b/src/aind_slims_api/models/user.py @@ -0,0 +1,28 @@ +"""Contains a model for a user.""" + +from typing import Optional + +from pydantic import Field + +from aind_slims_api.models.base import SlimsBaseModel + + +# TODO: Tighten this up once users are more commonly used +class SlimsUser(SlimsBaseModel): + """Model for user information in SLIMS + + Examples + -------- + >>> from aind_slims_api.core import SlimsClient + >>> client = SlimsClient() + >>> user = client.fetch_model(SlimsUser, username="LKim") + """ + + username: str = Field(..., alias="user_userName") + first_name: Optional[str] = Field("", alias="user_firstName") + last_name: Optional[str] = Field("", alias="user_lastName") + full_name: Optional[str] = Field("", alias="user_fullName") + email: Optional[str] = Field("", alias="user_email") + pk: int = Field(..., alias="user_pk") + + _slims_table = "User" diff --git a/src/aind_slims_api/models/utils.py b/src/aind_slims_api/models/utils.py new file mode 100644 index 0000000..3efd9ae --- /dev/null +++ b/src/aind_slims_api/models/utils.py @@ -0,0 +1,30 @@ +"""Utility functions and classes for working with slims models. +""" + +from typing import Optional + +from pydantic.fields import FieldInfo + + +class UnitSpec: + """Used in type annotation metadata to specify units""" + + units: tuple[str] + preferred_unit: Optional[str] = None + + def __init__(self, *args, preferred_unit=None): + """Set list of acceptable units from args, and preferred_unit""" + self.units = args + if len(self.units) == 0: + raise ValueError("One or more units must be specified") + if preferred_unit is None: + self.preferred_unit = self.units[0] + + +def _find_unit_spec(field: FieldInfo) -> UnitSpec | None: + """Given a Pydantic FieldInfo, find the UnitSpec in its metadata""" + metadata = field.metadata + for m in metadata: + if isinstance(m, UnitSpec): + return m + return None diff --git a/src/aind_slims_api/mouse.py b/src/aind_slims_api/mouse.py deleted file mode 100644 index ad3a439..0000000 --- a/src/aind_slims_api/mouse.py +++ /dev/null @@ -1,73 +0,0 @@ -"""Contains a model for the mouse content, and a method for fetching it""" - -import logging -from typing import Annotated - -from pydantic import Field, BeforeValidator, ValidationError - -from aind_slims_api.core import SlimsBaseModel, SlimsClient, UnitSpec, SLIMSTABLES - -logger = logging.getLogger() - - -class SlimsMouseContent(SlimsBaseModel): - """Model for an instance of the Mouse ContentType""" - - baseline_weight_g: Annotated[float | None, UnitSpec("g")] = Field( - ..., alias="cntn_cf_baselineWeight" - ) - point_of_contact: str | None = Field(..., alias="cntn_cf_scientificPointOfContact") - water_restricted: Annotated[bool, BeforeValidator(lambda x: x or False)] = Field( - ..., alias="cntn_cf_waterRestricted" - ) - barcode: str = Field(..., alias="cntn_barCode") - pk: int = Field(..., alias="cntn_pk") - - _slims_table: SLIMSTABLES = "Content" - - # TODO: Include other helpful fields (genotype, gender...) - - # pk: callable - # cntn_fk_category: SlimsColumn - # cntn_fk_contentType: SlimsColumn - # cntn_barCode: SlimsColumn - # cntn_id: SlimsColumn - # cntn_cf_contactPerson: SlimsColumn - # cntn_status: SlimsColumn - # cntn_fk_status: SlimsColumn - # cntn_fk_user: SlimsColumn - # cntn_cf_fk_fundingCode: SlimsColumn - # cntn_cf_genotype: SlimsColumn - # cntn_cf_labtracksId: SlimsColumn - # cntn_cf_parentBarcode: SlimsColumn - - -def fetch_mouse_content( - client: SlimsClient, - mouse_name: str, -) -> SlimsMouseContent | dict | None: - """Fetches mouse information for a mouse with labtracks id {mouse_name}""" - mice = client.fetch( - "Content", - cntp_name="Mouse", - cntn_barCode=mouse_name, - ) - - if len(mice) > 0: - mouse_details = mice[0] - if len(mice) > 1: - logger.warning( - f"Warning, Multiple mice in SLIMS with barcode " - f"{mouse_name}, using pk={mouse_details.cntn_pk.value}" - ) - else: - logger.warning("Warning, Mouse not in SLIMS") - return None - - try: - mouse = SlimsMouseContent.model_validate(mouse_details) - except ValidationError as e: - logger.error(f"SLIMS data validation failed, {repr(e)}") - return mouse_details.json_entity - - return mouse diff --git a/src/aind_slims_api/types.py b/src/aind_slims_api/types.py new file mode 100644 index 0000000..64fab16 --- /dev/null +++ b/src/aind_slims_api/types.py @@ -0,0 +1,19 @@ +"""Common types for the SLIMS API. +""" + +from typing import Literal + +# List of slims tables manually accessed, there are many more +SLIMS_TABLES = Literal[ + "Attachment", + "Project", + "Content", + "ContentEvent", + "Unit", + "Result", + "Test", + "User", + "Groups", + "Instrument", + "Unit", +] diff --git a/src/aind_slims_api/user.py b/src/aind_slims_api/user.py deleted file mode 100644 index 74a6994..0000000 --- a/src/aind_slims_api/user.py +++ /dev/null @@ -1,55 +0,0 @@ -"""Contains a model for a user, and a method for fetching it""" - -import logging -from typing import Optional - -from pydantic import Field, ValidationError - -from aind_slims_api.core import SlimsBaseModel, SlimsClient - -logger = logging.getLogger() - - -# TODO: Tighten this up once users are more commonly used -class SlimsUser(SlimsBaseModel): - """Model for user information in SLIMS""" - - username: str = Field(..., alias="user_userName") - first_name: Optional[str] = Field("", alias="user_firstName") - last_name: Optional[str] = Field("", alias="user_lastName") - full_name: Optional[str] = Field("", alias="user_fullName") - email: Optional[str] = Field("", alias="user_email") - pk: int = Field(..., alias="user_pk") - - _slims_table: str = "User" - - -def fetch_user( - client: SlimsClient, - username: str, -) -> SlimsUser | dict | None: - """Fetches user information for a user with username {username}""" - users = client.fetch( - "User", - user_userName=username, - ) - - if len(users) > 0: - user_details = users[0] - if len(users) > 1: - logger.warning( - f"Warning, Multiple users in SLIMS with " - f"username {username}, " - f"using pk={user_details.pk()}" - ) - else: - logger.warning("Warning, User not in SLIMS") - return - - try: - user = SlimsUser.model_validate(user_details) - except ValidationError as e: - logger.error(f"SLIMS data validation failed, {repr(e)}") - return user_details.json_entity - - return user diff --git a/src/aind_slims_api/write_models.py b/src/aind_slims_api/write_models.py new file mode 100644 index 0000000..e606a28 --- /dev/null +++ b/src/aind_slims_api/write_models.py @@ -0,0 +1,49 @@ +"""Model writing utilities adding records to the SLIMS database. + +Notes +----- +- Here due to restructuring of models, will likely be deprecated in the future. +""" + +import logging + +from aind_slims_api.core import SlimsClient +from aind_slims_api.models import ( + SlimsBehaviorSession, + SlimsInstrument, + SlimsMouseContent, + SlimsUser, +) + +logger = logging.getLogger() + + +def write_behavior_session_content_events( + client: SlimsClient, + mouse: SlimsMouseContent, + instrument: SlimsInstrument, + trainers: list[SlimsUser], + *behavior_sessions: SlimsBehaviorSession, +) -> list[SlimsBehaviorSession]: + """Writes behavior sessions to the SLIMS database. + + Notes + ----- + - Here due to restructuring of models, will likely be deprecated in the + future. + """ + trainer_pks = [trainer.pk for trainer in trainers] + logger.debug(f"Trainer pks: {trainer_pks}") + added = [] + for behavior_session in behavior_sessions: + updated = behavior_session.model_copy( + update={ + "mouse_pk": mouse.pk, + "instrument": instrument.pk, + "trainers": trainer_pks, + }, + ) + logger.debug(f"Resolved behavior session: {updated}") + added.append(client.add_model(updated)) + + return added diff --git a/tests/resources/example_fetch_attachments_response.json_entity.json b/tests/resources/example_fetch_attachments_response.json_entity.json new file mode 100644 index 0000000..2cd17d6 --- /dev/null +++ b/tests/resources/example_fetch_attachments_response.json_entity.json @@ -0,0 +1,336 @@ +[{ + "canDelete": true, + "canUpdate": true, + "columns": [ + { + "datatype": "STRING", + "editable": true, + "hidden": false, + "name": "attm_name", + "position": 0, + "title": "Name", + "value": "rig323_EPHYS1_OPTO_2024-02-12.json" + }, + { + "datatype": "STRING", + "editable": true, + "hidden": false, + "name": "attm_uniqueIdentifier", + "position": 1, + "title": "Unique identifier", + "value": "21387bbf-d1a5-4c26-86c1-2bce06f21c98" + }, + { + "datatype": "FOREIGN_KEY", + "displayField": "attp_name", + "displayValue": null, + "editable": false, + "foreignDisplayColumn": "attp_name", + "foreignTable": "AttachmentType", + "hidden": true, + "name": "attm_fk_attachmentType", + "position": 2, + "title": "Type", + "value": null + }, + { + "datatype": "FOREIGN_KEY", + "displayField": "user_userName", + "displayValue": null, + "editable": true, + "foreignDisplayColumn": "user_userName", + "foreignTable": "User", + "hidden": false, + "name": "attm_fk_user", + "position": 3, + "title": "User", + "value": null + }, + { + "datatype": "FOREIGN_KEY", + "displayField": "grps_groupName", + "displayValue": null, + "editable": true, + "foreignDisplayColumn": "grps_groupName", + "foreignTable": "Groups", + "hidden": false, + "name": "attm_fk_group", + "position": 4, + "title": "Group", + "value": null + }, + { + "datatype": "BOOLEAN", + "editable": true, + "hidden": false, + "name": "attm_mainReport", + "position": 5, + "title": "Main report", + "value": false + }, + { + "datatype": "INTEGER", + "editable": true, + "hidden": false, + "name": "attm_file_filesize", + "position": 6, + "title": "Size", + "value": 21125 + }, + { + "datatype": "DATE", + "dateFormat": "MM/dd/yyyy HH:mm:ss", + "editable": true, + "hidden": false, + "name": "attm_file_date_created", + "position": 7, + "subType": "datetime", + "timeZone": "America/Los_Angeles", + "title": "Date", + "value": 1711642546857 + }, + { + "datatype": "BOOLEAN", + "editable": true, + "hidden": false, + "name": "attm_isDirectory", + "position": 8, + "title": "Directory", + "value": false + }, + { + "datatype": "INTEGER", + "editable": true, + "hidden": false, + "name": "attm_linkCount", + "position": 9, + "title": "# of links to this attachment", + "value": 1 + }, + { + "datatype": "BOOLEAN", + "editable": true, + "hidden": false, + "name": "attm_currentlyLinked", + "position": 10, + "title": "Currently linked", + "value": null + }, + { + "datatype": "STRING", + "editable": true, + "hidden": false, + "name": "attm_createdBy", + "position": 11, + "title": "Created by", + "value": "LKim" + }, + { + "datatype": "DATE", + "dateFormat": "MM/dd/yyyy HH:mm:ss", + "editable": true, + "hidden": false, + "name": "attm_createdOn", + "position": 12, + "subType": "datetime", + "timeZone": "America/Los_Angeles", + "title": "Created on", + "value": 1711642546906 + }, + { + "datatype": "STRING", + "editable": true, + "hidden": false, + "name": "attm_modifiedBy", + "position": 13, + "title": "Modified by", + "value": "LKim" + }, + { + "datatype": "DATE", + "dateFormat": "MM/dd/yyyy HH:mm:ss", + "editable": true, + "hidden": false, + "name": "attm_modifiedOn", + "position": 14, + "subType": "datetime", + "timeZone": "America/Los_Angeles", + "title": "Modified on", + "value": 1711642546906 + }, + { + "datatype": "STRING", + "editable": true, + "hidden": false, + "name": "attm_ecm3Url", + "position": 15, + "title": "ECM Server URI", + "value": null + }, + { + "datatype": "FOREIGN_KEY", + "displayField": "user_userName", + "displayValue": null, + "editable": true, + "foreignDisplayColumn": "user_userName", + "foreignTable": "User", + "hidden": false, + "name": "attm_fk_ecm3UploadUser", + "position": 16, + "title": "ECM Upload user", + "value": 7 + }, + { + "datatype": "ENUM", + "editable": true, + "hidden": false, + "name": "attm_ecm3UploadStatus", + "position": 17, + "title": "ECM Upload Status", + "value": null + }, + { + "datatype": "STRING", + "editable": true, + "hidden": true, + "name": "grps_groupName", + "position": 18, + "title": "Name", + "value": null + }, + { + "datatype": "STRING", + "editable": true, + "hidden": true, + "name": "attm_externalId", + "position": 19, + "title": "attm_externalId", + "value": null + }, + { + "datatype": "ENUM", + "editable": true, + "hidden": true, + "name": "attm_analysisRole", + "position": 20, + "title": "attm_analysisRole", + "value": null + }, + { + "datatype": "STRING", + "editable": true, + "hidden": true, + "name": "attm_file_filename", + "position": 21, + "title": "File name", + "value": "rig323_EPHYS1_OPTO_2024-02-12.json" + }, + { + "datatype": "INTEGER", + "editable": true, + "hidden": true, + "name": "attm_pk", + "position": 22, + "title": "attm_pk", + "value": 21 + }, + { + "datatype": "STRING", + "editable": true, + "hidden": true, + "name": "attp_name", + "position": 23, + "title": "Name", + "value": null + }, + { + "datatype": "BOOLEAN", + "editable": true, + "hidden": true, + "name": "attm_isRemote", + "position": 24, + "title": "Remote", + "value": false + }, + { + "datatype": "BOOLEAN", + "editable": true, + "hidden": true, + "name": "attm_isMachineData", + "position": 25, + "title": "Machine data", + "value": false + }, + { + "datatype": "ENUM", + "displayValue": "Not indexable", + "editable": true, + "hidden": true, + "name": "attm_indexingStatus", + "position": 26, + "title": "Indexing status", + "value": "NOT_INDEXABLE" + }, + { + "datatype": "STRING", + "editable": true, + "hidden": true, + "name": "attm_ecm3Id", + "position": 27, + "title": "ECM ID", + "value": null + }, + { + "datatype": "STRING", + "editable": true, + "hidden": true, + "name": "attm_hash", + "position": 28, + "title": "Hash", + "value": "1e1eb28f644f2c51c209371eec79bd21" + }, + { + "datatype": "STRING", + "editable": true, + "hidden": true, + "name": "attm_path", + "position": 29, + "title": "Path", + "value": "ff/f7c/b0b6f9b42788ac937769afea8be/rig323_EPHYS1_OPTO_2024-02-12.json" + }, + { + "datatype": "STRING", + "editable": true, + "hidden": true, + "name": "user_userName", + "position": 30, + "title": "User name", + "value": null + } + ], + "links": [ + { + "href": "https://aind-test.us.slims.agilent.com/slimsrest/rest/Attachment/21", + "rel": "self" + }, + { + "href": "https://aind-test.us.slims.agilent.com/slimsrest/rest/User/7", + "rel": "attm_fk_ecm3UploadUser" + }, + { + "href": "https://aind-test.us.slims.agilent.com/slimsrest/rest/AttachmentLink?atln_fk_attachment=21", + "rel": "-atln_fk_attachment" + }, + { + "href": "https://aind-test.us.slims.agilent.com/slimsrest/rest/attachment/Attachment/21", + "rel": "attachments" + }, + { + "href": "https://aind-test.us.slims.agilent.com/slimsrest/rest/repo/21", + "rel": "contents" + } + ], + "pk": 21, + "tableName": "Attachment" + } +] \ No newline at end of file diff --git a/tests/test_behavior_session.py b/tests/test_behavior_session.py index 71644a3..6ab0d66 100644 --- a/tests/test_behavior_session.py +++ b/tests/test_behavior_session.py @@ -3,23 +3,18 @@ import json import os import unittest -from pathlib import Path -from unittest.mock import MagicMock, patch from datetime import datetime +from pathlib import Path +from unittest.mock import MagicMock, patch, call from slims.internal import Record from aind_slims_api.core import SlimsClient -from aind_slims_api.mouse import ( - SlimsMouseContent, -) -from aind_slims_api.user import SlimsUser -from aind_slims_api.instrument import SlimsInstrument -from aind_slims_api.behavior_session import ( - fetch_behavior_session_content_events, - write_behavior_session_content_events, - SlimsBehaviorSessionContentEvent, -) +from aind_slims_api.models.behavior_session import SlimsBehaviorSession +from aind_slims_api.models.instrument import SlimsInstrument +from aind_slims_api.models.mouse import SlimsMouseContent +from aind_slims_api.models.user import SlimsUser +from aind_slims_api.write_models import write_behavior_session_content_events RESOURCES_DIR = Path(os.path.dirname(os.path.realpath(__file__))) / "resources" @@ -30,9 +25,9 @@ class TestBehaviorSession(unittest.TestCase): example_client: SlimsClient example_response: list[Record] example_mouse_response: list[Record] - example_behavior_sessions: list[SlimsBehaviorSessionContentEvent] + example_behavior_sessions: list[SlimsBehaviorSession] example_mouse: SlimsMouseContent - example_write_sessions_response: list[Record] + example_write_sessions_response: Record example_instrument: SlimsInstrument example_trainer: SlimsUser @@ -80,7 +75,7 @@ def setUpClass(cls): cntn_cf_baselineWeight=None, ) cls.example_behavior_sessions = [ - SlimsBehaviorSessionContentEvent( + SlimsBehaviorSession( cnvn_cf_notes="Test notes", cnvn_cf_taskStage="Test stage", cnvn_cf_task="Test task", @@ -104,56 +99,38 @@ def setUpClass(cls): def test_fetch_behavior_session_content_events_success(self, mock_fetch: MagicMock): """Test fetch_behavior_session_content_events when successful""" mock_fetch.return_value = self.example_response - validated, unvalidated = fetch_behavior_session_content_events( - self.example_client, self.example_mouse + validated = self.example_client.fetch_models( + SlimsBehaviorSession, + mouse_pk=self.example_mouse.pk, + sort="date", ) - ret_entities = [item.json_entity for item in validated] + [ - item["json_entity"] for item in unvalidated - ] self.assertEqual( [item.json_entity for item in self.example_response], - ret_entities, + [item.json_entity for item in validated], ) @patch("slims.slims.Slims.fetch") - def test_fetch_behavior_session_content_events_success_unvalidated( + def test_fetch_behavior_session_content_events_success_sort_list( self, mock_fetch: MagicMock ): """Test fetch_behavior_session_content_events when successful""" mock_fetch.return_value = self.example_response - validated, unvalidated = fetch_behavior_session_content_events( - self.example_client, self.example_mouse_response[0].json_entity + validated = self.example_client.fetch_models( + SlimsBehaviorSession, + mouse_pk=self.example_mouse.pk, + sort=["date"], ) - ret_entities = [item.json_entity for item in validated] + [ - item["json_entity"] for item in unvalidated - ] self.assertEqual( [item.json_entity for item in self.example_response], - ret_entities, + [item.json_entity for item in validated], ) - @patch("slims.slims.Slims.fetch") - def test_fetch_behavior_session_content_events_failure_none( - self, mock_fetch: MagicMock - ): - """Test fetch_behavior_session_content_events when supplied with None""" - mock_fetch.return_value = self.example_response - with self.assertRaises(ValueError): - fetch_behavior_session_content_events(self.example_client, None) - - @patch("slims.slims.Slims.fetch") - def test_fetch_behavior_session_content_events_failure_bad_value( - self, mock_fetch: MagicMock - ): - """Test fetch_behavior_session_content_events when supplied with None""" - mock_fetch.return_value = self.example_response - with self.assertRaises(ValueError): - fetch_behavior_session_content_events(self.example_client, 1) - + @patch("aind_slims_api.core.logger") @patch("slims.slims.Slims.add") def test_write_behavior_session_content_events_success( self, mock_add: MagicMock, + mock_log_info: MagicMock ): """Test write_behavior_session_content_events success""" mock_add.return_value = self.example_write_sessions_response @@ -164,10 +141,11 @@ def test_write_behavior_session_content_events_success( [self.example_trainer], *self.example_behavior_sessions, ) - self.assertTrue( - all((item.mouse_pk == self.example_mouse.pk for item in added)) - ) + self.assertTrue(all((item.mouse_pk == self.example_mouse.pk for item in added))) self.assertTrue(len(added) == len(self.example_behavior_sessions)) + mock_log_info.assert_has_calls( + [call.info('SLIMS Add: ContentEvent/79')] + ) if __name__ == "__main__": diff --git a/tests/test_core.py b/tests/test_core.py index 6ac5923..803e130 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -7,11 +7,13 @@ from pathlib import Path from unittest.mock import MagicMock, patch +from requests import Response from slims.criteria import conjunction, equals from slims.internal import Record, _SlimsApiException -from aind_slims_api.core import SlimsClient -from aind_slims_api.unit import SlimsUnit +from aind_slims_api.core import SlimsAttachment, SlimsClient +from aind_slims_api.exceptions import SlimsRecordNotFound +from aind_slims_api.models.unit import SlimsUnit RESOURCES_DIR = Path(os.path.dirname(os.path.realpath(__file__))) / "resources" @@ -19,6 +21,12 @@ class TestSlimsClient(unittest.TestCase): """Tests methods in SlimsClient class""" + example_client: SlimsClient + example_fetch_unit_response: list[Record] + example_fetch_mouse_response: list[Record] + example_fetch_user_response: list[Record] + example_fetch_attachment_response: list[Record] + @classmethod def setUpClass(cls): """Sets up class by downloading responses""" @@ -42,6 +50,9 @@ def get_response(attribute_name: str): cls.example_fetch_unit_response = get_response("example_fetch_unit_response") cls.example_fetch_mouse_response = get_response("example_fetch_mouse_response") cls.example_fetch_user_response = get_response("example_fetch_user_response") + cls.example_fetch_attachment_response = get_response( + "example_fetch_attachments_response.json_entity" + ) def test_rest_link(self): """Tests rest_link method with both queries and no queries.""" @@ -175,6 +186,22 @@ def test_update_failure( mock_update.assert_not_called() mock_log.assert_not_called() + @patch("slims.slims.Slims.fetch_by_pk") + @patch("logging.Logger.info") + @patch("slims.internal.Record.update") + def test_update_model_no_pk( + self, + mock_update: MagicMock, + mock_log: MagicMock, + mock_fetch_by_pk: MagicMock, + ): + """Tests update method when a failure occurs""" + mock_fetch_by_pk.return_value = None + with self.assertRaises(ValueError): + self.example_client.update_model(SlimsUnit.model_construct(pk=None)) + mock_update.assert_not_called() + mock_log.assert_not_called() + @patch("logging.Logger.info") @patch("slims.slims.Slims.add") def test_add_model(self, mock_slims_add: MagicMock, mock_log: MagicMock): @@ -213,6 +240,79 @@ def test_update_model( self.assertEqual(updated_model, returned_model) mock_log.assert_called_once_with("SLIMS Update: Unit/31") + @patch("slims.slims.Slims.fetch") + def test_fetch_model_no_records(self, mock_slims_fetch: MagicMock): + """Tests fetch_user method""" + mock_slims_fetch.return_value = [] + with self.assertRaises(SlimsRecordNotFound): + self.example_client.fetch_model(SlimsUnit) + + def test_fetch_attachments(self): + """Tests fetch_attachments method success.""" + # slims_api is dynamically added to slims client + assert len(self.example_fetch_attachment_response) == 1 + with patch.object( + self.example_client.db.slims_api, + "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, + ) + ) + attachments = self.example_client.fetch_attachments( + unit, + ) + assert len(attachments) == 1 + + def test_fetch_attachment_content(self): + """Tests fetch_attachment_content method success.""" + # slims_api is dynamically added to slims client + with patch.object( + self.example_client.db.slims_api, + "get", + return_value=Response(), + ): + self.example_client.fetch_attachment_content( + SlimsAttachment( + attm_name="test", + attm_pk=1, + ) + ) + + @patch("logging.Logger.error") + def test__validate_model_invalid_model(self, mock_log: MagicMock): + """Tests _validate_model method with one invalid model and one valid + one. + """ + valid_data = deepcopy(self.example_fetch_unit_response[0].json_entity) + invalid_data = deepcopy(self.example_fetch_unit_response[0].json_entity) + invalid_data["columns"][0]["value"] = 1 + validated = self.example_client._validate_models( + SlimsUnit, + [ + Record( + json_entity=valid_data, + slims_api=self.example_client.db.slims_api, + ), + Record( + json_entity=invalid_data, + slims_api=self.example_client.db.slims_api, + ), + ], + ) + assert len(validated) == 1 + assert mock_log.call_count == 1 + + 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") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_instrument.py b/tests/test_instrument.py index f6c8d0d..44fe5ce 100644 --- a/tests/test_instrument.py +++ b/tests/test_instrument.py @@ -3,16 +3,13 @@ import json import os import unittest -from copy import deepcopy from pathlib import Path from unittest.mock import MagicMock, patch from slims.internal import Record from aind_slims_api.core import SlimsClient -from aind_slims_api.instrument import ( - fetch_instrument_content, -) +from aind_slims_api.models.instrument import SlimsInstrument RESOURCES_DIR = Path(os.path.dirname(os.path.realpath(__file__))) / "resources" @@ -38,46 +35,19 @@ def setUpClass(cls): ) ] - @patch("logging.Logger.warning") @patch("slims.slims.Slims.fetch") def test_fetch_content_success( self, mock_fetch: MagicMock, - mock_log_warn: MagicMock, ): """Test fetch_instrument_content when successful and multiple are returned from fetch """ mock_fetch.return_value = self.example_response + self.example_response - response = fetch_instrument_content(self.example_client, "323_EPHYS1_OPTO") - self.assertEqual(response.json_entity, self.example_response[0].json_entity) - self.assertTrue(mock_log_warn.called) - - @patch("slims.slims.Slims.fetch") - def test_fetch_fail( - self, - mock_fetch: MagicMock, - ): - """Test fetch_instrument_content when invalid instrument name is given.""" - mock_fetch.return_value = [] - response = fetch_instrument_content( - self.example_client, "Hopefully not a valid instrument name right?" + response = self.example_client.fetch_model( + SlimsInstrument, name="323_EPHYS1_OPTO" ) - self.assertTrue(response is None) - - @patch("slims.slims.Slims.fetch") - def test_fetch_unvalidated_success( - self, - mock_fetch: MagicMock, - ): - """Test fetch_instrument_content when unvalidated instrument data - returned. - """ - bad_return = deepcopy(self.example_response[0]) - bad_return.nstr_pk.value = "burrito" - mock_fetch.return_value = [bad_return, bad_return] - response = fetch_instrument_content(self.example_client, "323_EPHYS1_OPTO") - self.assertTrue(isinstance(response, dict)) + self.assertEqual(response.json_entity, self.example_response[0].json_entity) if __name__ == "__main__": diff --git a/tests/test_mouse.py b/tests/test_mouse.py index 9a8f72d..6c84c96 100644 --- a/tests/test_mouse.py +++ b/tests/test_mouse.py @@ -5,12 +5,11 @@ import unittest from pathlib import Path from unittest.mock import MagicMock, patch -from copy import deepcopy from slims.internal import Record from aind_slims_api.core import SlimsClient -from aind_slims_api.mouse import fetch_mouse_content +from aind_slims_api.models.mouse import SlimsMouseContent RESOURCES_DIR = Path(os.path.dirname(os.path.realpath(__file__))) / "resources" @@ -18,6 +17,9 @@ class TestMouse(unittest.TestCase): """Tests top level methods in mouse module""" + example_client: SlimsClient + example_fetch_mouse_response: list[Record] + @classmethod def setUpClass(cls): """Load json files of expected responses from slims""" @@ -36,55 +38,12 @@ def setUpClass(cls): def test_fetch_mouse_content_success(self, mock_fetch: MagicMock): """Test fetch_mouse_content when successful""" mock_fetch.return_value = self.example_fetch_mouse_response - mouse_details = fetch_mouse_content(self.example_client, mouse_name="123456") - self.assertEqual( - self.example_fetch_mouse_response[0].json_entity, mouse_details.json_entity + mouse_details = self.example_client.fetch_model( + SlimsMouseContent, barcode="123456" ) - - @patch("logging.Logger.warning") - @patch("slims.slims.Slims.fetch") - def test_fetch_mouse_content_no_mouse( - self, mock_fetch: MagicMock, mock_log_warn: MagicMock - ): - """Test fetch_mouse_content when no mouse is returned""" - mock_fetch.return_value = [] - mouse_details = fetch_mouse_content(self.example_client, mouse_name="12") - self.assertIsNone(mouse_details) - mock_log_warn.assert_called_with("Warning, Mouse not in SLIMS") - - @patch("logging.Logger.warning") - @patch("slims.slims.Slims.fetch") - def test_fetch_mouse_content_many_mouse( - self, mock_fetch: MagicMock, mock_log_warn: MagicMock - ): - """Test fetch_mouse_content when too many mice are returned""" - mock_fetch.return_value = [ - self.example_fetch_mouse_response[0], - self.example_fetch_mouse_response[0], - ] - mouse_details = fetch_mouse_content(self.example_client, mouse_name="123456") self.assertEqual( self.example_fetch_mouse_response[0].json_entity, mouse_details.json_entity ) - mock_log_warn.assert_called_with( - "Warning, Multiple mice in SLIMS with barcode 123456, using pk=3038" - ) - - @patch("logging.Logger.error") - @patch("slims.slims.Slims.fetch") - def test_fetch_mouse_content_validation_fail( - self, mock_fetch: MagicMock, mock_log_error: MagicMock - ): - """Test fetch_mouse when successful""" - wrong_return = deepcopy(self.example_fetch_mouse_response) - wrong_return[0].cntn_cf_waterRestricted.value = 14 - mock_fetch.return_value = wrong_return - mouse_info = fetch_mouse_content(self.example_client, mouse_name="123456") - self.assertEqual( - self.example_fetch_mouse_response[0].json_entity, - mouse_info, - ) - mock_log_error.assert_called() if __name__ == "__main__": diff --git a/tests/test_slimsmodel.py b/tests/test_slimsmodel.py index a3ff69b..53e1925 100644 --- a/tests/test_slimsmodel.py +++ b/tests/test_slimsmodel.py @@ -1,13 +1,14 @@ """ Tests the generic SlimsBaseModel""" +import unittest from datetime import datetime from typing import Annotated -import unittest from pydantic import Field -from slims.internal import Record, Column +from slims.internal import Column, Record -from aind_slims_api.core import SlimsBaseModel, UnitSpec +from aind_slims_api.core import SlimsBaseModel +from aind_slims_api.models.utils import UnitSpec class TestSlimsModel(unittest.TestCase): diff --git a/tests/test_user.py b/tests/test_user.py index 1ea9a81..06e940b 100644 --- a/tests/test_user.py +++ b/tests/test_user.py @@ -5,12 +5,11 @@ import unittest from pathlib import Path from unittest.mock import MagicMock, patch -from copy import deepcopy from slims.internal import Record from aind_slims_api.core import SlimsClient -from aind_slims_api.user import fetch_user +from aind_slims_api.models.user import SlimsUser RESOURCES_DIR = Path(os.path.dirname(os.path.realpath(__file__))) / "resources" @@ -18,6 +17,9 @@ class TestUser(unittest.TestCase): """Tests top level methods in user module""" + example_client: SlimsClient + example_fetch_user_response: list[Record] + @classmethod def setUpClass(cls): """Load json files of expected responses from slims""" @@ -36,62 +38,11 @@ def setUpClass(cls): def test_fetch_user_content_success(self, mock_fetch: MagicMock): """Test fetch_user when successful""" mock_fetch.return_value = self.example_fetch_user_response - user_info = fetch_user(self.example_client, username="PersonA") - self.assertEqual( - self.example_fetch_user_response[0].json_entity, - user_info.json_entity, - ) - - @patch("logging.Logger.error") - @patch("slims.slims.Slims.fetch") - def test_fetch_user_content_validation_fail( - self, mock_fetch: MagicMock, mock_log_error: MagicMock - ): - """Test fetch_user when successful""" - wrong_return = deepcopy(self.example_fetch_user_response) - wrong_return[0].user_userName.value = 14 - mock_fetch.return_value = wrong_return - user_info = fetch_user(self.example_client, username="PersonA") - self.assertEqual( - self.example_fetch_user_response[0].json_entity, - user_info, - ) - mock_log_error.assert_called() - - @patch("logging.Logger.warning") - @patch("slims.slims.Slims.fetch") - def test_fetch_user_content_no_user( - self, mock_fetch: MagicMock, mock_log_warn: MagicMock - ): - """Test fetch_user when no user is returned""" - mock_fetch.return_value = [] - user_info = fetch_user(self.example_client, username="PersonX") - self.assertIsNone(user_info) - mock_log_warn.assert_called_with("Warning, User not in SLIMS") - - @patch("logging.Logger.warning") - @patch("slims.slims.Slims.fetch") - def test_fetch_user_content_many_users( - self, mock_fetch: MagicMock, mock_log_warn: MagicMock - ): - """Test fetch_user_content when too many users are returned""" - mocked_response = [ - self.example_fetch_user_response[0], - self.example_fetch_user_response[0], - ] - mock_fetch.return_value = mocked_response - username = "PersonA" - user_info = fetch_user(self.example_client, username=username) + user_info = self.example_client.fetch_model(SlimsUser, username="PersonA") self.assertEqual( self.example_fetch_user_response[0].json_entity, user_info.json_entity, ) - expected_warning = ( - f"Warning, Multiple users in SLIMS with " - f"username {username}, " - f"using pk={mocked_response[0].pk()}" - ) - mock_log_warn.assert_called_with(expected_warning) if __name__ == "__main__":