Generic Client (#10)

* Initial add

* Update dependencies

* refactor: expand SlimsBaseModel and client

* feat: add mouse model and fetch method

* feat: add user model and fetch method

* Feat client unit tests (#11)

* docs: set line length to 88

* tests: adds test coverage

---------

Co-authored-by: jtyoung84 <[email protected]>

.flake8

@@ -4,3 +4,4 @@ exclude =
     __pycache__,
     build
 max-complexity = 10
+max-line-length = 88

.github/workflows/test_and_lint.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [ '3.8', '3.9', '3.10' ]
+        python-version: [ '3.10', '3.11', '3.12' ]
     steps:
     - uses: actions/checkout@v3
     - name: Set up Python ${{ matrix.python-version }}

.gitignore

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

doc_template/source/conf.py

@@ -1,12 +1,15 @@
 """Configuration file for the Sphinx documentation builder."""
+
 #
 # For the full list of built-in configuration values, see the documentation:
 # https://www.sphinx-doc.org/en/master/usage/configuration.html
 
+from datetime import date
+
 # -- Path Setup --------------------------------------------------------------
-from os.path import dirname, abspath
+from os.path import abspath, dirname
 from pathlib import Path
-from datetime import date
+
 from aind_slims_api import __version__ as package_version
 
 INSTITUTE_NAME = "Allen Institute for Neural Dynamics"

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 name = "aind-slims-api"
 description = "Generated from aind-library-template"
 license = {text = "MIT"}
-requires-python = ">=3.7"
+requires-python = ">=3.10"
 authors = [
     {name = "Allen Institute for Neural Dynamics"}
 ]
@@ -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]
@@ -37,8 +40,8 @@ where = ["src"]
 version = {attr = "aind_slims_api.__version__"}
 
 [tool.black]
-line-length = 79
-target_version = ['py36']
+line-length = 88
+target_version = ['py310']
 exclude = '''
 
 (
@@ -71,7 +74,7 @@ exclude_lines = [
 fail_under = 100
 
 [tool.isort]
-line_length = 79
+line_length = 88
 profile = "black"
 
 [tool.interrogate]

src/aind_slims_api/__init__.py

@@ -1,2 +1,9 @@
 """Init package"""
+
 __version__ = "0.0.0"
+
+from aind_slims_api.configuration import AindSlimsApiSettings
+
+config = AindSlimsApiSettings()
+
+from aind_slims_api.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,135 @@
+"""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
+"""
+
+import logging
+from functools import lru_cache
+from typing import Literal, Optional
+
+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
+
+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: Optional[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 as e:
+            # TODO: Add better error handling
+            #  Let's just raise error for the time being
+            raise e
+
+        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(f'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,33 @@
+"""Contains a model for the mouse content, and a method for fetching it"""
+
+import logging
+from typing import Optional
+
+from aind_slims_api.core import SlimsClient
+
+logger = logging.getLogger()
+
+
+def fetch_mouse_content(
+    client: SlimsClient,
+    mouse_name: str,
+) -> Optional[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.value}"
+            )
+    else:
+        logger.warning("Warning, Mouse not in SLIMS")
+        mouse_details = None
+
+    return None if mouse_details is None else mouse_details.json_entity

src/aind_slims_api/user.py

@@ -0,0 +1,33 @@
+"""Contains a model for a user, and a method for fetching it"""
+
+import logging
+from typing import Optional
+
+from aind_slims_api.core import SlimsClient
+
+logger = logging.getLogger()
+
+
+def fetch_user(
+    client: SlimsClient,
+    username: str,
+) -> Optional[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 {[u.json_entity for u in users]}, "
+                f"using pk={user_details.pk()}"
+            )
+    else:
+        logger.warning("Warning, User not in SLIMS")
+        user_details = None
+
+    return None if user_details is None else user_details.json_entity