Add iacuc function to utils

src/aind_metadata_mapper/utils.py

@@ -19,6 +19,7 @@
 INSTRUMENT_BASE_URL = "http://aind-metadata-service/api/v2/instrument"
 PROCEDURES_BASE_URL = "http://aind-metadata-service/api/v2/procedures"
 SUBJECT_BASE_URL = "http://aind-metadata-service/api/v2/subject"
+LABTRACKS_SUBJECT_BASE_URL = "http://aind-metadata-service/api/v2/labtracks/subject"
 
 
 def normalize_utc_timezone(dt: str) -> str:
@@ -173,6 +174,46 @@ def get_procedures(subject_id: str, base_url: str = PROCEDURES_BASE_URL, timeout
         return None
 
 
+def get_iacuc_protocol(subject_id: str | int, base_url: str = LABTRACKS_SUBJECT_BASE_URL) -> Optional[str]:
+    """Fetch a subject's current IACUC protocol number from LabTracks.
+
+    LabTracks is the regulatory source of truth for which protocol a mouse is on. The
+    metadata service exposes the LabTracks subject record, which includes a
+    ``protocol_number`` field (e.g. "2414") joined directly from the LabTracks
+    ``IacucProtocol`` table.
+
+    Note that, in the future, this information will also be tracked in DataVerse.
+    This function could be updated to pull from there in the future if desired.
+
+    Parameters
+    ----------
+    subject_id : str or int
+        The subject ID to query. An int is cast to str automatically.
+    base_url : str
+        Base URL for the LabTracks subject endpoint. Defaults to LABTRACKS_SUBJECT_BASE_URL.
+
+    Returns
+    -------
+    Optional[str]
+        The IACUC protocol number (e.g. "2414"), or None if the subject is not in
+        LabTracks, has no associated protocol, or the request fails.
+    """
+    try:
+        subject_id = str(subject_id)
+        # This endpoint takes subject_id as a query parameter, not a path segment.
+        url = f"{base_url}?subject_id={subject_id}"
+        records = metadata_service_helper(url)
+        if not records:
+            logger.warning(f"Could not fetch LabTracks subject {subject_id}")
+            return None
+        records = records if isinstance(records, list) else [records]
+        protocol_number = records[0].get("protocol_number")
+        return str(protocol_number) if protocol_number else None
+    except Exception as e:
+        logger.warning(f"Unexpected error fetching IACUC protocol for subject {subject_id}: {e}")
+        return None
+
+
 def get_intended_measurements(
     subject_id: str,
     base_url: str = "http://aind-metadata-service/intended_measurements",

tests/test_utils.py

@@ -23,6 +23,7 @@
     check_existing_instrument,
     check_instrument_id,
     ensure_timezone,
+    get_iacuc_protocol,
     get_instrument,
     get_intended_measurements,
     get_procedures,
@@ -639,5 +640,63 @@ def input_func(prompt):
         self.assertTrue(any("Help" in str(call) for call in mock_print.call_args_list))
 
 
+class TestGetIacucProtocol(unittest.TestCase):
+    """Tests for get_iacuc_protocol function."""
+
+    @patch("aind_metadata_mapper.utils.metadata_service_helper")
+    def test_returns_protocol_number_field(self, mock_helper):
+        """The protocol_number field (joined from the IacucProtocol table) is returned."""
+        mock_helper.return_value = [{"group_name": "Exp-ND-01-001-2414", "protocol_number": "2414"}]
+        self.assertEqual(get_iacuc_protocol("818908"), "2414")
+
+    @patch("aind_metadata_mapper.utils.metadata_service_helper")
+    def test_accepts_int_subject_id(self, mock_helper):
+        """An int subject_id is cast to str and handled."""
+        mock_helper.return_value = [{"protocol_number": "2414"}]
+        self.assertEqual(get_iacuc_protocol(818908), "2414")
+
+    @patch("aind_metadata_mapper.utils.metadata_service_helper")
+    def test_returns_protocol_when_group_name_lacks_number(self, mock_helper):
+        """A breeding-group name with no trailing number still resolves via protocol_number."""
+        mock_helper.return_value = [{"group_name": "Pdyn-IRES-Cre;Oi9(ND)", "protocol_number": "2453"}]
+        self.assertEqual(get_iacuc_protocol("x"), "2453")
+
+    @patch("aind_metadata_mapper.utils.metadata_service_helper")
+    def test_numeric_protocol_number_is_coerced_to_str(self, mock_helper):
+        """A non-string protocol_number is coerced to str."""
+        mock_helper.return_value = [{"protocol_number": 2414}]
+        self.assertEqual(get_iacuc_protocol("x"), "2414")
+
+    @patch("aind_metadata_mapper.utils.metadata_service_helper")
+    def test_null_protocol_number_returns_none(self, mock_helper):
+        """A record with a null protocol_number yields None."""
+        mock_helper.return_value = [{"group_name": "Practice Mice", "protocol_number": None}]
+        self.assertIsNone(get_iacuc_protocol("x"))
+
+    @patch("aind_metadata_mapper.utils.metadata_service_helper")
+    def test_missing_protocol_number_returns_none(self, mock_helper):
+        """A record without a protocol_number yields None."""
+        mock_helper.return_value = [{}]
+        self.assertIsNone(get_iacuc_protocol("x"))
+
+    @patch("aind_metadata_mapper.utils.metadata_service_helper")
+    def test_no_records_returns_none(self, mock_helper):
+        """No records (helper returns None) yields None."""
+        mock_helper.return_value = None
+        self.assertIsNone(get_iacuc_protocol("x"))
+
+    @patch("aind_metadata_mapper.utils.metadata_service_helper")
+    def test_dict_error_body_returns_none(self, mock_helper):
+        """A non-list error body (e.g. {"detail": "Not Found"}) yields None."""
+        mock_helper.return_value = {"detail": "Not Found"}
+        self.assertIsNone(get_iacuc_protocol("x"))
+
+    @patch("aind_metadata_mapper.utils.metadata_service_helper")
+    def test_exception_returns_none(self, mock_helper):
+        """An unexpected error is caught and yields None."""
+        mock_helper.side_effect = ValueError("boom")
+        self.assertIsNone(get_iacuc_protocol("x"))
+
+
 if __name__ == "__main__":
     unittest.main()