add contents generator

Author: bernt-matthias

bioblend/_tests/TestGalaxyFolders.py

@@ -54,6 +54,21 @@ def test_show_folder_contents_limit(self):
         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")

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,
@@ -85,6 +86,40 @@ def show_folder(
         }
         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'
+
+        :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]:
         """
         Marks the folder with the given ``id`` as `deleted` (or removes the