bug(fcsfile): fix recursion when file has no spill string

Author: gegnew

cellengine/resources/fcs_file.py

@@ -35,7 +35,8 @@ class FcsFile(DataClassMixin):
     sample_name: Optional[str] = field(default=ReadOnly(optional=True))  # type: ignore
     size: int = field(default=ReadOnly())  # type: ignore
     _spill_string: Optional[str] = field(
-        metadata=config(field_name="spillString"), default=None
+        metadata=config(field_name="spillString"),
+        default=None,
     )
 
     def __repr__(self):
@@ -197,6 +198,10 @@ def plot(
 
     def get_file_internal_compensation(self) -> Compensation:
         """Get the file-internal Compensation."""
+        if not self.has_file_internal_comp:
+            raise ValueError(
+                f"FCS File '{self._id}' does not have an internal compensation."
+            )
         return Compensation.from_spill_string(self.spill_string)
 
     @property
@@ -223,12 +228,11 @@ def events(self, events):
 
     @property
     def spill_string(self):
-        if self._spill_string:
-            return self._spill_string
-        else:
-            ss = ce.APIClient().get_fcs_file(self.experiment_id, self._id).spill_string
-            self._spill_string = ss
-            return ss
+        if not self._spill_string and self.has_file_internal_comp:
+            self._spill_string = ce.APIClient().get_fcs_file(
+                self.experiment_id, self._id, as_dict=True
+            )["spillString"]
+        return self._spill_string or ""
 
     def get_events(
         self, inplace: bool = False, destination=None, **kwargs

cellengine/utils/api_client/APIClient.py

@@ -320,9 +320,29 @@ def get_fcs_files(self, experiment_id, as_dict=False) -> List[FcsFile]:
             return fcs_files
         return [FcsFile.from_dict(fcs_file) for fcs_file in fcs_files]
 
+    # fmt: off
+    @overload
     def get_fcs_file(
-        self, experiment_id, _id=None, name=None, as_dict=False
-    ) -> FcsFile:
+        self,
+        experiment_id: str,
+        _id: str = None,
+        name: str = None,
+        as_dict: bool = True,
+    ) -> Dict[str, Any]: ...
+
+    @overload
+    def get_fcs_file(
+        self,
+        experiment_id: str,
+        _id: str = None,
+        name: str = None,
+        as_dict: bool = False,
+    ) -> FcsFile: ...
+    # fmt: on
+
+    def get_fcs_file(
+        self, experiment_id: str, _id: str = None, name: str = None, as_dict=False
+    ):
         _id = _id or self._get_id_by_name(name, "fcsfiles", experiment_id)
         fcs_file = self._get(
             f"{self.base_url}/experiments/{experiment_id}/fcsfiles/{_id}"

tests/unit/resources/test_fcsfile.py

@@ -1,12 +1,14 @@
-import os
+from io import BufferedReader, BytesIO
 import json
+import os
+
 from fcsparser.api import FCSParser
 from pandas.core.frame import DataFrame
+import pytest
 import responses
-from io import BufferedReader, BytesIO
 
-from cellengine.resources.fcs_file import FcsFile
 from cellengine.resources.compensation import Compensation
+from cellengine.resources.fcs_file import FcsFile
 
 
 EXP_ID = "5d38a6f79fae87499999a74b"
@@ -79,6 +81,32 @@ def test_gets_file_internal_compensation(ENDPOINT_BASE, client, fcs_files, spill
     assert type(comp) == Compensation
 
 
+@responses.activate
+def test_throws_correct_error_when_no_file_internal_compensation(
+    ENDPOINT_BASE, client, fcs_files
+):
+    # Given: An FcsFile with no spill string
+    file_data = fcs_files[0]
+    file_data["spillString"] = None
+    file_data["hasFileInternalComp"] = False
+    file = FcsFile.from_dict(file_data)
+    expected_response = fcs_files[0].copy()
+    responses.add(
+        responses.GET,
+        f"{ENDPOINT_BASE}/experiments/{EXP_ID}/fcsfiles/{file._id}",
+        json=expected_response,
+    )
+
+    # When:
+    with pytest.raises(ValueError) as err:
+        comp = file.get_file_internal_compensation()
+        # Then:
+    assert (
+        err.value.args[0]
+        == f"FCS File '{file._id}' does not have an internal compensation."
+    )
+
+
 def test_parse_fcs_file():
     events_body = open("tests/data/Acea - Novocyte.fcs", "rb")
     parser = FCSParser.from_data(events_body.read())