diff --git a/src/aind_slims_api/__init__.py b/src/aind_slims_api/__init__.py index a7d8a94..12f17df 100644 --- a/src/aind_slims_api/__init__.py +++ b/src/aind_slims_api/__init__.py @@ -7,3 +7,28 @@ config = AindSlimsApiSettings() from aind_slims_api.core import SlimsClient # noqa + + +def testmod(**testmod_kwargs): + """ + Run doctests for the module, configured to ignore exception details and + normalize whitespace. Also sets logging level to DEBUG. + + Accepts kwargs to pass to doctest.testmod(). + + Add to modules to run doctests when run as a script: + .. code-block:: text + if __name__ == "__main__": + from npc_io import testmod + testmod() + + """ + import logging + import doctest + + logging.basicConfig(level=logging.DEBUG) + + _ = testmod_kwargs.setdefault( + "optionflags", doctest.NORMALIZE_WHITESPACE | doctest.ELLIPSIS + ) + doctest.testmod(**testmod_kwargs) diff --git a/src/aind_slims_api/behavior_session.py b/src/aind_slims_api/behavior_session.py index d271caa..f9d90d6 100644 --- a/src/aind_slims_api/behavior_session.py +++ b/src/aind_slims_api/behavior_session.py @@ -3,7 +3,7 @@ """ import logging -from typing import Any +from typing import ClassVar from datetime import datetime from pydantic import Field @@ -17,7 +17,16 @@ class SlimsBehaviorSessionContentEvent(SlimsBaseModel): - """Model for an instance of the Behavior Session ContentEvent""" + """Model for an instance of the Behavior Session ContentEvent + + Examples + -------- + >>> from aind_slims_api.core import SlimsClient + >>> from aind_slims_api.mouse import SlimsMouseContent + >>> client = SlimsClient() + >>> mouse = client.fetch_model(SlimsMouseContent, barcode="00000000") + >>> behavior_sessions = client.fetch_models(SlimsBehaviorSessionContentEvent, mouse_pk=mouse.pk, sort=["date"]) + """ pk: int | None = Field(default=None, alias="cnvn_pk") mouse_pk: int | None = Field( @@ -36,55 +45,12 @@ class SlimsBehaviorSessionContentEvent(SlimsBaseModel): ) software_version: str | None = Field( default=None, alias="cnvn_cf_softwareVersion") - date: datetime | None = Field(..., alias="cnvn_cf_scheduledDate") + date: datetime | None = Field(default=None, 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: SlimsMouseContent, -) -> list[SlimsBehaviorSessionContentEvent]: - """Fetches behavior sessions for a mouse with labtracks id {mouse_name} - - Returns - ------- - list: - Validated SlimsBehaviorSessionContentEvent objects - """ - return client.fetch_models( - SlimsBehaviorSessionContentEvent, - cnvn_fk_content=_resolve_pk(mouse), - cnvt_name="Behavior Session", - sort=["cnvn_cf_scheduledDate"], - ) + _base_fetch_filters: ClassVar[dict[str, int | str]] = { + "cnvt_name": "Behavior Session", + } def write_behavior_session_content_events( @@ -101,18 +67,14 @@ def write_behavior_session_content_events( - 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] + 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, + "mouse_pk": mouse.pk, + "instrument": instrument.pk, "trainers": trainer_pks, }, ) @@ -120,3 +82,9 @@ def write_behavior_session_content_events( added.append(client.add_model(updated)) return added + + +if __name__ == "__main__": + from aind_slims_api import testmod + + testmod() 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 6754f73..180f21e 100644 --- a/src/aind_slims_api/core.py +++ b/src/aind_slims_api/core.py @@ -22,14 +22,14 @@ from pydantic.fields import FieldInfo import logging from typing import ( - Any, Generator, Callable, Literal, Optional, Sequence, Type, TypeVar, + ClassVar, Literal, Optional, Type, + TypeVar, ) from requests import Response from slims.slims import Slims, _SlimsApiException from slims.internal import ( Column as SlimsColumn, Record as SlimsRecord, - Attachment as SlimsAttachment, ) from slims.criteria import Criterion, conjunction, equals @@ -49,6 +49,7 @@ "User", "Groups", "Instrument", + "Unit", ] @@ -56,7 +57,7 @@ class UnitSpec: """Used in type annotation metadata to specify units""" units: list[str] - preferred_unit: str = None + preferred_unit: Optional[str] = None def __init__(self, *args, preferred_unit=None): """Set list of acceptable units from args, and preferred_unit""" @@ -94,11 +95,10 @@ class MyModel(SlimsBaseModel): Datetime fields will be serialized to an integer ms timestamp """ - pk: int = None - json_entity: dict = None - attachments: Optional[Callable[[], Sequence[SlimsAttachment]]] = None + pk: Optional[int] = None + json_entity: Optional[dict] = None _slims_table: SLIMSTABLES - _base_fetch_filters: dict[str, str | int] # use for fetch_models, fetch_model + _base_fetch_filters: ClassVar[dict[str, str | int]] = {} # use for fetch_models, fetch_model @field_validator("*", mode="before") def _validate(cls, value, info: ValidationInfo): @@ -119,11 +119,6 @@ def _validate(cls, value, info: ValidationInfo): ) raise ValueError(msg) return value.value - elif info.field_name in ("attachments", ): - if isinstance(value, dict): - return None - else: - return value else: return value @@ -145,54 +140,13 @@ def _serialize(self, field, info): # TODO: Add links - need Record.json_entity['links']['self'] # TODO: Add Table - need Record.json_entity['tableName'] - # def fetch_attachments(self) -> Sequence['Attachment']: - # """Fetches all attachments for this record""" - # if not self.attachments: - # logger.debug("Initialized without attachments.") - # return [] - # validated = [] - # for attachment in self.attachments(): - # try: - # validated.append(Attachment.model_validate(attachment)) - # except ValidationError as e: - # logger.error(f"SLIMS data validation failed, {repr(e)}") - # return validated - - # def fetch_attachments_content(self) -> \ - # Generator[Response, None, None]: - # """Fetches the content of all attachments for this record and returns - # them as dictionaries. - - # Notes - # ----- - # - Assumes that the attachments are json - # - Should this actually be text and not json? - # """ - # for attachment in self.fetch_attachments(): - # yield attachment.fetch_content() - class Attachment(SlimsBaseModel): pk: int = Field(..., alias="attm_pk") name: str = Field(..., alias="attm_name") - # slims_api: Slims _slims_table: SLIMSTABLES = "Attachment" - # class Config: - # arbitrary_types_allowed = True - - # def fetch_content(self) -> Response: - # """Fetches the content of this attachment""" - # return self.slims_api.get(f"repo/{self.pk}") - - # attm_pk: SlimsColumn - # attm_fk_content: SlimsColumn - # attm_fk_user: SlimsColumn - # attm_fk_attachmentType: SlimsColumn - # attm_fk_status: SlimsColumn - # attm_fk_file - SlimsBaseModelTypeVar = TypeVar("SlimsBaseModelTypeVar", bound=SlimsBaseModel) @@ -272,8 +226,8 @@ def resolve_model_alias( ) -> str: """Given a SlimsBaseModel object, resolve its pk to the actual value""" for field_name, field_info in model.model_fields.items(): - if field_info.alias == attr_name: - return field_name + 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}") @@ -315,14 +269,6 @@ def fetch_models( resolved_kwargs = deepcopy(model._base_fetch_filters) for name, value in kwargs.items(): resolved_kwargs[self.resolve_model_alias(model, name)] = value - # resolved_kwargs = { - # self.resolve_model_alias(model, name): value - # for name, value in kwargs.items() - # } - # for filter_name, filter_value in model._base_fetch_filters: - # if filter_name in resolved_kwargs: - - # resolved_kwargs[filter_name] = filter_value logger.debug("Resolved kwargs: %s", resolved_kwargs) resolved_sort: Optional[str | list[str]] = None if sort is not None: @@ -386,17 +332,6 @@ def fetch_attachments( def fetch_attachment_content(self, attachment: Attachment) -> Response: return self.db.slims_api.get(f"repo/{attachment.pk}") - # def fetch_attachments_contents( - # self, - # record: SlimsBaseModel, - # ) -> Generator[Response, None, None]: - # """Fetches all attachments for a record""" - # if not record.attachments: - # raise ValueError("Record initialized no attachments.") - - # for attachment in record.attachments(): - # yield attachment.slims_api.get(f"repo/{attachment.attm_pk}") - @lru_cache(maxsize=None) def fetch_pk(self, table: SLIMSTABLES, *args, **kwargs) -> int | None: """SlimsClient.fetch but returns the pk of the first returned record""" @@ -470,6 +405,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/instrument.py b/src/aind_slims_api/instrument.py index 10190dd..108b8f7 100644 --- a/src/aind_slims_api/instrument.py +++ b/src/aind_slims_api/instrument.py @@ -1,7 +1,6 @@ """Contains a model for the instrument content, and a method for fetching it""" import logging -from typing import Any from pydantic import Field @@ -17,70 +16,21 @@ class SlimsInstrument(SlimsBaseModel): -------- >>> from aind_slims_api.core import SlimsClient >>> client = SlimsClient() - >>> instrument = client.fetch_model(SlimsInstrument, nstr_name="323_EPHYS1_OPTO") + >>> instrument = client.fetch_model(SlimsInstrument, name="323_EPHYS1_OPTO") """ - name: str = Field(..., alias="nstr_name") + name: str = Field( + ..., + alias="nstr_name", + description="The name of the instrument", + ) 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 - - if __name__ == "__main__": - import doctest - import logging - - logging.basicConfig(level=logging.DEBUG) + from aind_slims_api import testmod - doctest.testmod( - optionflags=( - doctest.IGNORE_EXCEPTION_DETAIL | doctest.NORMALIZE_WHITESPACE - ) - ) \ No newline at end of file + testmod() diff --git a/src/aind_slims_api/mouse.py b/src/aind_slims_api/mouse.py index ad3a439..41828c9 100644 --- a/src/aind_slims_api/mouse.py +++ b/src/aind_slims_api/mouse.py @@ -1,7 +1,7 @@ """Contains a model for the mouse content, and a method for fetching it""" import logging -from typing import Annotated +from typing import Annotated, ClassVar from pydantic import Field, BeforeValidator, ValidationError @@ -11,7 +11,14 @@ class SlimsMouseContent(SlimsBaseModel): - """Model for an instance of the Mouse ContentType""" + """Model for an instance of the Mouse ContentType + + 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" @@ -24,6 +31,9 @@ class SlimsMouseContent(SlimsBaseModel): pk: int = Field(..., alias="cntn_pk") _slims_table: SLIMSTABLES = "Content" + _base_fetch_filters: ClassVar[dict[str, int | str]] = { + "cntp_name": "Mouse", + } # TODO: Include other helpful fields (genotype, gender...) @@ -71,3 +81,9 @@ def fetch_mouse_content( return mouse_details.json_entity return mouse + + +if __name__ == "__main__": + from aind_slims_api import testmod + + testmod() diff --git a/src/aind_slims_api/unit.py b/src/aind_slims_api/unit.py index 73c2efc..20a8709 100644 --- a/src/aind_slims_api/unit.py +++ b/src/aind_slims_api/unit.py @@ -5,7 +5,7 @@ from pydantic import Field -from aind_slims_api.core import SlimsBaseModel +from aind_slims_api.core import SlimsBaseModel, SLIMSTABLES logger = logging.getLogger() @@ -17,4 +17,4 @@ class SlimsUnit(SlimsBaseModel): abbreviation: Optional[str] = Field("", alias="unit_abbreviation") pk: int = Field(..., alias="unit_pk") - _slims_table: str = "Unit" + _slims_table: SLIMSTABLES = "Unit" diff --git a/src/aind_slims_api/user.py b/src/aind_slims_api/user.py index 74a6994..bc94a31 100644 --- a/src/aind_slims_api/user.py +++ b/src/aind_slims_api/user.py @@ -1,18 +1,19 @@ """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 +from pydantic import Field -logger = logging.getLogger() +from aind_slims_api.core import SlimsBaseModel, SLIMSTABLES # TODO: Tighten this up once users are more commonly used class SlimsUser(SlimsBaseModel): - """Model for user information in SLIMS""" + """Model for user information in SLIMS + + >>> from aind_slims_api.core import SlimsClient + >>> client = SlimsClient() + >>> user = client.fetch_model(SlimsUser, "LisaK") + """ username: str = Field(..., alias="user_userName") first_name: Optional[str] = Field("", alias="user_firstName") @@ -21,35 +22,10 @@ class SlimsUser(SlimsBaseModel): 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 + _slims_table: SLIMSTABLES = "User" + + +if __name__ == "__main__": + from aind_slims_api import testmod + + testmod()