Feat 140 slims client

.github/workflows/ci.yml

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

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.8-slim
+FROM python:3.9-slim
 WORKDIR /app
 
 # Install FreeTDS and dependencies

pyproject.toml

@@ -9,7 +9,7 @@ license = {text = "MIT"}
 authors = [
     {name = "Allen Institute for Neural Dynamics"}
 ]
-requires-python = ">=3.7"
+requires-python = ">=3.9"
 classifiers = [
     "Programming Language :: Python :: 3"
 ]
@@ -38,7 +38,8 @@ server = [
     'office365-rest-python-client==2.4.1',
     'fastapi',
     'uvicorn[standard]',
-    'python-dateutil'
+    'python-dateutil',
+    'slims-python-api==6.8.0',
 ]
 
 client = [
@@ -54,7 +55,7 @@ readme = {file = ["README.md"]}
 
 [tool.black]
 line-length = 79
-target_version = ['py36']
+target_version = ['py39']
 exclude = '''
 
 (

src/aind_metadata_service/labtracks/client.py

@@ -349,8 +349,11 @@ def _map_housing(
         room_id = None if room_id is None or int(room_id) < 0 else room_id
         cage_id = None if cage_id is None or int(cage_id) < 0 else cage_id
 
-        return Housing.construct(room_id=room_id,
-                                 cage_id=cage_id) if room_id is not None or cage_id is not None else None
+        return (
+            Housing.construct(room_id=room_id, cage_id=cage_id)
+            if room_id is not None or cage_id is not None
+            else None
+        )
 
     def map_response_to_subject(self, results: List[dict]) -> List[Subject]:
         """

src/aind_metadata_service/sharepoint/nsb2023/mapping.py

@@ -2955,8 +2955,7 @@ def burr_hole_info(self, burr_hole_num: int) -> BurrHoleInfo:
                 alternating_current=self.aind_inj1_alternating_time,
                 inj_duration=self.aind_inj1_ionto_time,
                 inj_volume=self._map_burr_hole_volume(
-                    vol=self.aind_inj1volperdepth,
-                    dv=coordinate_depth
+                    vol=self.aind_inj1volperdepth, dv=coordinate_depth
                 ),
                 fiber_implant_depth=self.aind_fiber_implant1_dv,
             )
@@ -2979,8 +2978,7 @@ def burr_hole_info(self, burr_hole_num: int) -> BurrHoleInfo:
                 alternating_current=self.aind_inj2_alternating_time,
                 inj_duration=self.aind_inj2_ionto_time,
                 inj_volume=self._map_burr_hole_volume(
-                    vol=self.aind_inj2volperdepth,
-                    dv=coordinate_depth
+                    vol=self.aind_inj2volperdepth, dv=coordinate_depth
                 ),
                 fiber_implant_depth=self.aind_fiber_implant2_dv,
             )
@@ -3003,8 +3001,7 @@ def burr_hole_info(self, burr_hole_num: int) -> BurrHoleInfo:
                 alternating_current=self.aind_inj3_alternating_time,
                 inj_duration=self.aind_inj3_ionto_time,
                 inj_volume=self._map_burr_hole_volume(
-                    vol=self.aind_inj3volperdepth,
-                    dv=coordinate_depth
+                    vol=self.aind_inj3volperdepth, dv=coordinate_depth
                 ),
                 fiber_implant_depth=self.aind_fiber_implant3_d_x00,
             )
@@ -3027,8 +3024,7 @@ def burr_hole_info(self, burr_hole_num: int) -> BurrHoleInfo:
                 alternating_current=self.aind_inj4_alternating_time,
                 inj_duration=self.aind_inj4_ionto_time,
                 inj_volume=self._map_burr_hole_volume(
-                    vol=self.aind_inj4volperdepth,
-                    dv=coordinate_depth
+                    vol=self.aind_inj4volperdepth, dv=coordinate_depth
                 ),
                 fiber_implant_depth=self.aind_fiber_implant4_d_x00,
             )
@@ -3051,8 +3047,7 @@ def burr_hole_info(self, burr_hole_num: int) -> BurrHoleInfo:
                 alternating_current=self.aind_inj5_alternating_time,
                 inj_duration=self.aind_inj5_ionto_time,
                 inj_volume=self._map_burr_hole_volume(
-                    vol=self.aind_inj5volperdepth,
-                    dv=coordinate_depth
+                    vol=self.aind_inj5volperdepth, dv=coordinate_depth
                 ),
                 fiber_implant_depth=self.aind_fiber_implant5_d_x00,
             )
@@ -3075,8 +3070,7 @@ def burr_hole_info(self, burr_hole_num: int) -> BurrHoleInfo:
                 alternating_current=self.aind_inj6_alternating_time,
                 inj_duration=self.aind_inj6_ionto_time,
                 inj_volume=self._map_burr_hole_volume(
-                    vol=self.aind_inj6volperdepth,
-                    dv=coordinate_depth
+                    vol=self.aind_inj6volperdepth, dv=coordinate_depth
                 ),
                 fiber_implant_depth=self.aind_fiber_implant6_d_x00,
             )

src/aind_metadata_service/slims/__init__.py

@@ -0,0 +1 @@
+"""Package to connect to SLIMS db"""

src/aind_metadata_service/slims/client.py

@@ -0,0 +1,68 @@
+"""Module for slims client"""
+import logging
+
+from pydantic import BaseSettings, Extra, Field, SecretStr
+from slims.criteria import equals
+from slims.internal import Record
+from slims.slims import Slims
+
+from aind_metadata_service.slims.models import ContentsTableRow
+
+
+class SlimsSettings(BaseSettings):
+    """Configuration class. Mostly a wrapper around smartsheet.Smartsheet
+    class constructor arguments."""
+
+    username: str = Field(..., description="User name")
+    password: SecretStr = Field(..., description="Password")
+    host: str = Field(..., description="host")
+    db: str = Field(default="slims", description="Database")
+
+    class Config:
+        """Set env prefix and forbid extra fields."""
+
+        env_prefix = "SLIMS_"
+        extra = Extra.forbid
+
+
+class SlimsClient:
+    """Client to connect to slims db"""
+
+    def __init__(self, settings: SlimsSettings):
+        """Class constructor for slims client"""
+        self.settings = settings
+        self.client = Slims(
+            settings.db,
+            settings.host,
+            settings.username,
+            settings.password.get_secret_value(),
+        )
+
+    async def get_record(self, subject_id: str) -> Record:
+        """
+        Retrieve a record from the Contents Table
+        Parameters
+        ----------
+        subject_id : str
+          Labtracks id of subject to retrieve record for
+
+        Returns
+        -------
+        Record
+          A single slims Record
+
+        """
+        try:
+            content_record = self.client.fetch(
+                "Content",
+                equals(
+                    ContentsTableRow.__fields__["cntn_cf_labtracksId"].name,
+                    subject_id,
+                ),
+                start=0,
+                end=1,
+            )[0]
+            return content_record
+        except Exception as e:
+            logging.error(repr(e))
+            raise Exception(e)