use doctests, remove fetches

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)

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,22 +67,24 @@ 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,
             },
         )
         logger.debug(f"Resolved behavior session: {updated}")
         added.append(client.add_model(updated))
 
     return added
+
+
+if __name__ == "__main__":
+    from aind_slims_api import testmod
+
+    testmod()

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("")

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,14 +49,15 @@
     "User",
     "Groups",
     "Instrument",
+    "Unit",
 ]
 
 
 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,