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,72 @@ 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(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(self.folder["id"]))
+        assert len(f2) == 12
+        assert f2[0]["name"] == f"{self.name} 0"
+
+        # check non defaults
+        f2 = list(self.gi.folders.contents(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
+        # if called without contents=True an item_count is reported .. lets also check this
+        folder_info = self.gi.folders.show_folder(self.folder["id"], include_deleted=True)
+        assert folder_info["item_count"] == 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

@@ -5,6 +5,7 @@
 from typing import (
     Any,
     Dict,
+    Iterable,
     List,
     Literal,
     Optional,
@@ -45,7 +46,14 @@ 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]:
+    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 +64,61 @@ 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: Maximum number of contents to return (default: 10).
+          Only considered for contents=True.
+
+        :type offset: int
+        :param contents: Return contents from this specified position (default: 0).
+          Only considered for contents=True.
+
+        :type include_deleted: bool
+        :param include_deleted: Returns also deleted contents.
+          Only considered for contents=True.
+
         :rtype: dict
         :return: dictionary including details of the folder
         """
+        params = {
+            "limit": limit,
+            "offset": offset,
+            "include_deleted": include_deleted,
+        }
+        return self._get(id=folder_id, contents=contents, params=params)
+
+    def contents(
+        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'
 
-        return self._get(id=folder_id, contents=contents)
+        :type limit: int
+        :param limit: Batch size to be used internally (default: 10).
+
+        :type include_deleted: bool
+        :param include_deleted: Include also deleted contents.
+
+        :rtype: dict
+        :return: A generator for the folder contents
+        """
+        total_rows: Optional[int] = None
+        params = {
+            "limit": limit,
+            "offset": 0,
+            "include_deleted": include_deleted,
+        }
+
+        while total_rows is None or 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]:
         """