Generic Client

.gitignore

@@ -109,6 +109,7 @@ venv/
 ENV/
 env.bak/
 venv.bak/
+.conda
 
 # Spyder project settings
 .spyderproject

pyproject.toml

@@ -17,6 +17,9 @@ readme = "README.md"
 dynamic = ["version"]
 
 dependencies = [
+    'slims-python-api',
+    'pydantic',
+    'pydantic-settings'
 ]
 
 [project.optional-dependencies]
@@ -27,7 +30,7 @@ dev = [
     'interrogate',
     'isort',
     'Sphinx',
-    'furo'
+    'furo',
 ]
 
 [tool.setuptools.packages.find]

src/aind_slims_api/__init__.py

@@ -1,2 +1,9 @@
 """Init package"""
+
 __version__ = "0.0.0"
+
+from .configuration import AindSlimsApiSettings
+
+config = AindSlimsApiSettings()
+
+from .core import SlimsClient  # noqa

src/aind_slims_api/configuration.py

@@ -0,0 +1,16 @@
+""" Library Configuration model """
+
+from pydantic import SecretStr
+from pydantic_settings import BaseSettings
+
+
+class AindSlimsApiSettings(BaseSettings):
+    """Settings for SLIMS Client
+
+    Per pydantic-settings docs
+    https://docs.pydantic.dev/latest/concepts/pydantic_settings/
+    Loads slims credentials from environment variables if present"""
+
+    slims_url: str = "https://aind-test.us.slims.agilent.com/slimsrest/"
+    slims_username: str = ""
+    slims_password: SecretStr = ""

src/aind_slims_api/core.py

@@ -0,0 +1,136 @@
+"""Contents:
+
+Utilities for creating pydantic models for SLIMS data:
+    SlimsBaseModel - to be subclassed for SLIMS pydantic models
+    UnitSpec - To be included in a type annotation of a Quantity field
+
+SlimsClient - Basic wrapper around slims-python-api client with convenience
+    methods and integration with SlimsBaseModel subtypes
+"""
+
+from functools import lru_cache
+import logging
+from typing import Literal, Optional
+
+from slims.slims import Slims, _SlimsApiException
+from slims.internal import (
+    Record as SlimsRecord,
+)
+from slims.criteria import Criterion, conjunction, equals
+
+from aind_slims_api import config
+
+logger = logging.getLogger()
+
+# List of slims tables manually accessed, there are many more
+SLIMSTABLES = Literal[
+    "Project",
+    "Content",
+    "ContentEvent",
+    "Unit",
+    "Result",
+    "Test",
+    "User",
+    "Groups",
+]
+
+
+class SlimsClient:
+    """Wrapper around slims-python-api client with convenience methods"""
+
+    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: Slims = None
+
+        self.connect(
+            self.url,
+            username or config.slims_username,
+            password or config.slims_password.get_secret_value(),
+        )
+
+    def connect(self, url: str, username: str, password: str):
+        """Connect to the database"""
+        self.db = Slims(
+            "slims",
+            url,
+            username,
+            password,
+        )
+
+    def fetch(
+        self,
+        table: SLIMSTABLES,
+        *args,
+        sort: Optional[str | list[str]] = None,
+        start: Optional[int] = None,
+        end: Optional[int] = None,
+        **kwargs,
+    ) -> list[SlimsRecord]:
+        """Fetch from the SLIMS database
+
+        Args:
+            table (str): SLIMS table to query
+            sort (str | list[str], optional): Fields to sort by; e.g. date
+            start (int, optional):  The first row to return
+            end (int, optional): The last row to return
+            *args (Slims.criteria.Criterion): Optional criteria to apply
+            **kwargs (dict[str,str]): "field=value" filters
+
+        Returns:
+            records (list[SlimsRecord] | None): Matching records, if any
+        """
+        criteria = conjunction()
+        for arg in args:
+            if isinstance(arg, Criterion):
+                criteria.add(arg)
+
+        for k, v in kwargs.items():
+            criteria.add(equals(k, v))
+        try:
+            records = self.db.fetch(
+                table,
+                criteria,
+                sort=sort,
+                start=start,
+                end=end,
+            )
+        except _SlimsApiException:
+            raise
+            return None  # TODO: Raise or return empty list?
+
+        return records
+
+    @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"""
+        records = self.fetch(table, *args, **kwargs)
+        if len(records) > 0:
+            return records[0].pk()
+        else:
+            return None
+
+    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):
+        """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):
+        """Update a SLIMS record"""
+        record = self.db.fetch_by_pk(table, pk)
+        if record is None:
+            raise ValueError('No data in SLIMS "{table}" table for pk "{pk}"')
+        new_record = record.update(data)
+        logger.info(f"SLIMS Update: {table}/{pk}")
+        return new_record
+
+    def rest_link(self, table: SLIMSTABLES, **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()]
+        return base_url + "".join(queries)

src/aind_slims_api/mouse.py

@@ -0,0 +1,32 @@
+"""Contains a model for the mouse content, and a method for fetching it"""
+
+import logging
+
+from .core import SlimsClient
+
+logger = logging.getLogger()
+
+
+def fetch_mouse_content(
+    client: SlimsClient,
+    mouse_name: str,
+) -> dict:
+    """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}"
+            )
+    else:
+        logger.warning("Warning, Mouse not in SLIMS")
+        return
+
+    return mouse_details

src/aind_slims_api/user.py

@@ -0,0 +1,31 @@
+"""Contains a model for a user, and a method for fetching it"""
+
+import logging
+
+from .core import SlimsClient
+
+logger = logging.getLogger()
+
+
+def fetch_user(
+    client: SlimsClient,
+    username: str,
+) -> dict:
+    """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 {users}, using pk={user_details.pk}"
+            )
+    else:
+        logger.warning("Warning, User not in SLIMS")
+        return
+
+    return user_details

tests/test_slimsclient.py

@@ -0,0 +1,25 @@
+""" Tests the generic SlimsClient object"""
+
+import unittest
+
+# from aind_slims_api.configuration import AindSlimsApiSettings
+# from aind_slims_api.core import SlimsClient
+
+
+class TestSlimsClient(unittest.TestCase):
+    """Example Test Class"""
+
+    # def test_config(self):
+    #     config = AindSlimsApiSettings()
+
+    # def test_slims_client(self):
+    #     client = SlimsClient()
+
+    # @pytest.mark.parametrize("mouse_name", ["614173"])
+    # def test_waterlog(mouse_name):
+    #     wlclient = WaterlogSlimsClient()
+    #     wlclient.fetch_mouse_info(mouse_name)
+
+
+if __name__ == "__main__":
+    unittest.main()