Add limit, offset and include_deleted to show_folder

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"], limit=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"])

bioblend/_tests/TestGalaxyHistories.py

@@ -1,5 +1,4 @@
-"""
-"""
+""" """
 
 import os
 import shutil

bioblend/_tests/TestGalaxyTools.py

@@ -1,5 +1,4 @@
-"""
-"""
+""" """
 
 import os
 from typing import (

bioblend/_tests/test_util.py

@@ -1,5 +1,4 @@
-""" General support infrastructure not tied to any particular test.
-"""
+"""General support infrastructure not tied to any particular test."""
 
 import os
 import random

bioblend/galaxy/datasets/__init__.py

@@ -64,6 +64,13 @@ def show_dataset(self, dataset_id: str, hda_ldda: HdaLdda = "hda") -> Dict[str,
         }
         return self._get(id=dataset_id, params=params)
 
+    # def delete(self, dataset_ids, hda_ldda: HdaLdda = "hda", purge: bool = False) -> Dict[str, Any]:
+    #     datasets = []
+    #     for dataset_id in dataset_ids:
+    #         datasets.append({"src": hda_ldda, "id": dataset_id})
+    #     payload = {"datasets": datasets, "purge": purge}
+    #     return self._delete(payload=payload)
+
     def _initiate_download(
         self, dataset_id: str, stream_content: bool, require_ok_state: bool = True, maxwait: float = 12000
     ) -> Tuple[Dict[str, Any], str, Response]:

bioblend/galaxy/folders/__init__.py

@@ -2,12 +2,15 @@
 Contains possible interactions with the Galaxy library folders
 """
 
+import sys
 from typing import (
     Any,
     Dict,
+    Iterable,
     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,66 @@ 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 with 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,
+        limit: int = 10,
+        include_deleted: bool = False,
+    ) -> Iterable[Dict[str, Any]]:
         """
+        Iterate over folder contents.
+
+        :type folder_id: str
+        :param folder_id: the folder's encoded id, prefixed by 'F'
+
+        :type limit: int
+        :param limit: Batch size to be used internally.
 
-        return self._get(id=folder_id, contents=contents)
+        :type include_deleted: bool
+        :param include_deleted: Whether to include deleted items.
+
+        :rtype: dict
+        :return: A generator for the folder contents
+        """
+        total_rows = sys.maxsize
+        params = {
+            "limit": limit,
+            "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"] += limit
 
     def delete_folder(self, folder_id: str, undelete: bool = False) -> Dict[str, Any]:
         """