Add limit, offset and include_deleted to show_folder (#494)

bernt-matthias · nsoranzo · web-flow · commit 85315626aad8 · 2025-02-04T21:44:43.000Z
* add limit and offset to show_folder * add include_deleted * Simply test subfolder creation Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * Just use the upstream default values Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * Revert to loop and drop subfolders variable Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * add test for include_deleted * Simplify test Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * Simplify test Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * test content counts * fix test and add docs * total_rows became available only after galaxyproject/galaxy#10204 * add contents generator * Rename content to content_iter Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * Use sys.maxsize for iteration Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * overload show_folder * Apply suggestions from code review Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * Update bioblend/_tests/TestGalaxyFolders.py Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * revert accidental change * rewording Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com> * return is Iterator * rename limit * finish parameter renaming * more missing renaming --------- Co-authored-by: Nicola Soranzo <nicola.soranzo@gmail.com>
diff --git a/bioblend/_tests/TestGalaxyFolders.py b/bioblend/_tests/TestGalaxyFolders.py
@@ -3,7 +3,10 @@
     List,
 )
 
-from . import GalaxyTestBase
+from . import (
+    GalaxyTestBase,
+    test_util,
+)
 
 FOO_DATA = "foo\nbar\n"
 
@@ -36,6 +39,69 @@ def test_show_folder_contents(self):
         assert "metadata" in f2
         assert self.name == f2["metadata"]["folder_name"]
 
+    @test_util.skip_unless_galaxy("release_21.05")
+    def test_show_folder_contents_limit(self):
+        for i in range(12):
+            self.gi.folders.create_folder(self.folder["id"], f"{self.name} {i}")
+
+        # check defaults for limit and offset
+        f2 = self.gi.folders.show_folder(self.folder["id"], contents=True)
+        assert len(f2["folder_contents"]) == 10
+        assert f2["folder_contents"][0]["name"] == f"{self.name} 0"
+
+        # check non defaults
+        f2 = self.gi.folders.show_folder(self.folder["id"], contents=True, limit=1, offset=1)
+        assert len(f2["folder_contents"]) == 1
+        assert f2["folder_contents"][0]["name"] == f"{self.name} 1"
+
+    @test_util.skip_unless_galaxy("release_21.05")
+    def test_folder_contents_iter(self):
+        for i in range(12):
+            self.gi.folders.create_folder(self.folder["id"], f"{self.name} {i}")
+
+        # check defaults for limit and offset
+        f2 = list(self.gi.folders.contents_iter(self.folder["id"]))
+        assert len(f2) == 12
+        assert f2[0]["name"] == f"{self.name} 0"
+
+        # check non defaults
+        f2 = list(self.gi.folders.contents_iter(self.folder["id"], batch_size=1))
+        assert len(f2) == 12
+        assert f2[0]["name"] == f"{self.name} 0"
+
+    @test_util.skip_unless_galaxy("release_21.01")
+    def test_show_folder_contents_include_deleted(self):
+        history = self.gi.histories.create_history(name="Test History")
+        hda_id = self._test_dataset(history["id"])
+
+        # Create 2 library datasets into the library folder
+        ldda1 = self.gi.libraries.copy_from_dataset(
+            library_id=self.library["id"], dataset_id=hda_id, folder_id=self.folder["id"], message="Added HDA"
+        )
+        ldda2 = self.gi.libraries.copy_from_dataset(
+            library_id=self.library["id"], dataset_id=hda_id, folder_id=self.folder["id"], message="Added HDA"
+        )
+        folder_info = self.gi.folders.show_folder(self.folder["id"], contents=True)
+        assert len(folder_info["folder_contents"]) == 2
+        assert folder_info["folder_contents"][0]["type"] == "file"
+
+        # Delete the library datasets and check if include_deleted works
+        self.gi.libraries.delete_library_dataset(self.library["id"], ldda1["id"])
+        self.gi.libraries.delete_library_dataset(self.library["id"], ldda2["id"], purged=True)
+        folder_info = self.gi.folders.show_folder(self.folder["id"], contents=True, include_deleted=True)
+        # check if there are 2 contents and the number is correct
+        assert len(folder_info["folder_contents"]) == 2
+        assert folder_info["metadata"]["total_rows"] == 2
+
+        folder_info = self.gi.folders.show_folder(self.folder["id"], contents=True)
+        assert len(folder_info["folder_contents"]) == 0
+        assert folder_info["metadata"]["total_rows"] == 0
+        # show folders with contents=False does not respect include_deleted
+        folder_info = self.gi.folders.show_folder(self.folder["id"])
+        assert folder_info["item_count"] == 2
+
+        self.gi.histories.delete_history(history["id"])
+
     def test_delete_folder(self):
         self.sub_folder = self.gi.folders.create_folder(self.folder["id"], self.name)
         self.gi.folders.delete_folder(self.sub_folder["id"])
diff --git a/bioblend/galaxy/folders/__init__.py b/bioblend/galaxy/folders/__init__.py
@@ -2,12 +2,15 @@
 Contains possible interactions with the Galaxy library folders
 """
 
+import sys
 from typing import (
     Any,
     Dict,
+    Iterator,
     List,
     Literal,
     Optional,
+    overload,
     TYPE_CHECKING,
     Union,
 )
@@ -45,7 +48,31 @@ def create_folder(self, parent_folder_id: str, name: str, description: Optional[
             payload["description"] = description
         return self._post(payload=payload, id=parent_folder_id)
 
-    def show_folder(self, folder_id: str, contents: bool = False) -> Dict[str, Any]:
+    @overload
+    def show_folder(
+        self,
+        folder_id: str,
+        contents: Literal[False] = False,
+    ) -> Dict[str, Any]: ...
+
+    @overload
+    def show_folder(
+        self,
+        folder_id: str,
+        contents: Literal[True],
+        limit: int = 10,
+        offset: int = 0,
+        include_deleted: bool = False,
+    ) -> Dict[str, Any]: ...
+
+    def show_folder(
+        self,
+        folder_id: str,
+        contents: bool = False,
+        limit: int = 10,
+        offset: int = 0,
+        include_deleted: bool = False,
+    ) -> Dict[str, Any]:
         """
         Display information about a folder.
 
@@ -56,11 +83,63 @@ def show_folder(self, folder_id: str, contents: bool = False) -> Dict[str, Any]:
         :param contents: True to get the contents of the folder, rather
           than just the folder details.
 
+        :type limit: int
+        :param limit: When ``contents=True``, maximum number of items to return.
+
+        :type offset: int
+        :param contents: When ``contents=True``, number of items to skip. Return
+          contents starting from item offset+1.
+
+        :type include_deleted: bool
+        :param include_deleted: When ``contents=True``, whether to include
+          deleted items.
+
         :rtype: dict
-        :return: dictionary including details of the folder
+        :return: dictionary including details of the folder.
+          For contents=False the dict contains infos on the folder.
+          For contents=True the dict contains the keys "metadata" (a dict with
+          infos on the folder) and "folder_contents" (a list of dicts with info
+          on the childs).
+
+        Notes: For iterating over folder contents there is also contents_iter.
+        """
+        params = {
+            "limit": limit,
+            "offset": offset,
+            "include_deleted": include_deleted,
+        }
+        return self._get(id=folder_id, contents=contents, params=params)
+
+    def contents_iter(
+        self,
+        folder_id: str,
+        batch_size: int = 10,
+        include_deleted: bool = False,
+    ) -> Iterator[Dict[str, Any]]:
         """
+        Iterate over folder contents.
 
-        return self._get(id=folder_id, contents=contents)
+        :type folder_id: str
+        :param folder_id: the folder's encoded id, prefixed by 'F'
+
+        :type batch_size: int
+        :param batch_size: Batch size to be used internally.
+
+        :type include_deleted: bool
+        :param include_deleted: Whether to include deleted items.
+        """
+        total_rows = sys.maxsize
+        params = {
+            "limit": batch_size,
+            "offset": 0,
+            "include_deleted": include_deleted,
+        }
+
+        while params["offset"] <= total_rows:
+            chunk = self._get(id=folder_id, contents=True, params=params)
+            total_rows = chunk["metadata"]["total_rows"]
+            yield from chunk["folder_contents"]
+            params["offset"] += batch_size
 
     def delete_folder(self, folder_id: str, undelete: bool = False) -> Dict[str, Any]:
         """
