diff --git a/bioblend/_tests/TestGalaxyFolders.py b/bioblend/_tests/TestGalaxyFolders.py index d519fb8dc..4c1a8e98e 100644 --- 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/_tests/TestGalaxyHistories.py b/bioblend/_tests/TestGalaxyHistories.py index b1ae681b9..7d8ec1ee8 100644 --- a/bioblend/_tests/TestGalaxyHistories.py +++ b/bioblend/_tests/TestGalaxyHistories.py @@ -1,5 +1,4 @@ -""" -""" +""" """ import os import shutil diff --git a/bioblend/_tests/TestGalaxyTools.py b/bioblend/_tests/TestGalaxyTools.py index 2b3892a3c..b545f01f5 100644 --- a/bioblend/_tests/TestGalaxyTools.py +++ b/bioblend/_tests/TestGalaxyTools.py @@ -1,5 +1,4 @@ -""" -""" +""" """ import os from typing import ( diff --git a/bioblend/_tests/test_util.py b/bioblend/_tests/test_util.py index cdf1e8eeb..1cc47cd15 100644 --- a/bioblend/_tests/test_util.py +++ b/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 diff --git a/bioblend/galaxy/folders/__init__.py b/bioblend/galaxy/folders/__init__.py index 99f1c5bde..9c1fc79ad 100644 --- 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]: """