From b9b8dc99d169825afd48fc128fc80f60d77b86cc Mon Sep 17 00:00:00 2001 From: Alexander Piskun <13381981+bigcat88@users.noreply.github.com> Date: Fri, 11 Aug 2023 18:33:23 +0300 Subject: [PATCH] FileAPI: download_directory_as_zip (#73) Signed-off-by: Alexander Piskun --- .github/workflows/analysis-coverage.yml | 5 ++- CHANGELOG.md | 1 + nc_py_api/_session.py | 16 ++++++++ nc_py_api/files/files.py | 25 ++++++++++++ tests/files_test.py | 52 +++++++++++++++++++++++++ 5 files changed, 98 insertions(+), 1 deletion(-) diff --git a/.github/workflows/analysis-coverage.yml b/.github/workflows/analysis-coverage.yml index 853acd0d..7579bb6e 100644 --- a/.github/workflows/analysis-coverage.yml +++ b/.github/workflows/analysis-coverage.yml @@ -699,9 +699,12 @@ jobs: path: apps/app_ecosystem_v2 repository: cloud-py-api/app_ecosystem_v2 + - name: Patch base.php + if: ${{ startsWith(matrix.nextcloud, 'stable26') }} + run: patch -p 1 -i apps/app_ecosystem_v2/base_php.patch + - name: Install AppEcosystemV2 run: | - patch -p 1 -i apps/app_ecosystem_v2/base_php.patch php occ app:enable app_ecosystem_v2 cd nc_py_api coverage run --data-file=.coverage.ci_install tests/_install.py & diff --git a/CHANGELOG.md b/CHANGELOG.md index d9389f05..b84507b0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file. ### Added - APIs for enabling\disabling External Applications. +- FileAPI: `download_directory_as_zip` method. ### Changed diff --git a/nc_py_api/_session.py b/nc_py_api/_session.py index e4647127..6f747a80 100644 --- a/nc_py_api/_session.py +++ b/nc_py_api/_session.py @@ -150,6 +150,18 @@ def __del__(self): if hasattr(self, "adapter") and self.adapter: self.adapter.close() + def get_stream(self, path: str, params: Optional[dict] = None, **kwargs) -> Iterator[Response]: + return self._get_stream( + f"{quote(path)}?{urlencode(params, True)}" if params else quote(path), kwargs.get("headers", {}), **kwargs + ) + + def _get_stream(self, path_params: str, headers: dict, **kwargs) -> Iterator[Response]: + self.init_adapter() + timeout = kwargs.pop("timeout", self.cfg.options.timeout) + return self.adapter.stream( + "GET", f"{self.cfg.endpoint}{path_params}", headers=headers, timeout=timeout, **kwargs + ) + def ocs( self, method: str, @@ -296,6 +308,10 @@ def __init__(self, **kwargs): self.cfg = AppConfig(**kwargs) super().__init__(**kwargs) + def _get_stream(self, path_params: str, headers: dict, **kwargs) -> Iterator[Response]: + self.sign_request("GET", path_params, headers, None) + return super()._get_stream(path_params, headers, **kwargs) + def _ocs(self, method: str, path_params: str, headers: dict, data: Optional[bytes], **kwargs): self.sign_request(method, path_params, headers, data) return super()._ocs(method, path_params, headers, data, **kwargs) diff --git a/nc_py_api/files/files.py b/nc_py_api/files/files.py index afcb7360..fd9b823f 100644 --- a/nc_py_api/files/files.py +++ b/nc_py_api/files/files.py @@ -147,6 +147,31 @@ def download2stream(self, path: Union[str, FsNode], fp, **kwargs) -> None: else: raise TypeError("`fp` must be a path to file or an object with `write` method.") + def download_directory_as_zip( + self, path: Union[str, FsNode], local_path: Union[str, Path, None] = None, **kwargs + ) -> Path: + """Downloads a remote directory as zip archive. + + :param path: path to directory to download. + :param local_path: relative or absolute file path to save zip file. + :returns: Path to the saved zip archive. + + .. note:: This works only for directories, you should not use this to download a file. + """ + path = path.user_path if isinstance(path, FsNode) else path + with self._session.get_stream( + "/index.php/apps/files/ajax/download.php", params={"dir": path} + ) as response: # type: ignore + check_error(response.status_code, f"download_directory_as_zip: user={self._session.user}, path={path}") + result_path = local_path if local_path else os.path.basename(path) + with open( + result_path, + "wb", + ) as fp: + for data_chunk in response.iter_raw(chunk_size=kwargs.get("chunk_size", 4 * 1024 * 1024)): + fp.write(data_chunk) + return Path(result_path) + def upload(self, path: Union[str, FsNode], content: Union[bytes, str]) -> FsNode: """Creates a file with the specified content at the specified path. diff --git a/tests/files_test.py b/tests/files_test.py index 9451705d..ca7be5f7 100644 --- a/tests/files_test.py +++ b/tests/files_test.py @@ -1,4 +1,6 @@ import math +import os +import zipfile from datetime import datetime from io import BytesIO from random import choice, randbytes @@ -541,6 +543,56 @@ def test_fs_node_str(nc): nc.files.delete("test_file_name.txt") +@pytest.mark.parametrize("nc", NC_TO_TEST) +def test_download_as_zip(nc): + nc.files.makedirs("test_root_folder/test_subfolder", exist_ok=True) + try: + nc.files.mkdir("test_root_folder/test_subfolder2") + nc.files.upload("test_root_folder/0.txt", content="") + nc.files.upload("test_root_folder/1.txt", content="123") + nc.files.upload("test_root_folder/test_subfolder/0.txt", content="") + result = nc.files.download_directory_as_zip("test_root_folder") + try: + with zipfile.ZipFile(result, "r") as zip_ref: + assert zip_ref.filelist[0].filename == "test_root_folder/" + assert not zip_ref.filelist[0].file_size + assert zip_ref.filelist[1].filename == "test_root_folder/0.txt" + assert not zip_ref.filelist[1].file_size + assert zip_ref.filelist[2].filename == "test_root_folder/1.txt" + assert zip_ref.filelist[2].file_size == 3 + assert zip_ref.filelist[3].filename == "test_root_folder/test_subfolder/" + assert not zip_ref.filelist[3].file_size + assert zip_ref.filelist[4].filename == "test_root_folder/test_subfolder/0.txt" + assert not zip_ref.filelist[4].file_size + assert zip_ref.filelist[5].filename == "test_root_folder/test_subfolder2/" + assert not zip_ref.filelist[5].file_size + assert len(zip_ref.filelist) == 6 + finally: + os.remove(result) + result = nc.files.download_directory_as_zip("test_root_folder/test_subfolder", "2.zip") + try: + assert str(result) == "2.zip" + with zipfile.ZipFile(result, "r") as zip_ref: + assert zip_ref.filelist[0].filename == "test_subfolder/" + assert not zip_ref.filelist[0].file_size + assert zip_ref.filelist[1].filename == "test_subfolder/0.txt" + assert not zip_ref.filelist[1].file_size + assert len(zip_ref.filelist) == 2 + finally: + os.remove("2.zip") + result = nc.files.download_directory_as_zip("test_root_folder/test_subfolder2", "empty_folder.zip") + try: + assert str(result) == "empty_folder.zip" + with zipfile.ZipFile(result, "r") as zip_ref: + assert zip_ref.filelist[0].filename == "test_subfolder2/" + assert not zip_ref.filelist[0].file_size + assert len(zip_ref.filelist) == 1 + finally: + os.remove("empty_folder.zip") + finally: + nc.files.delete("test_root_folder") + + @pytest.mark.parametrize("nc", NC_TO_TEST[:1]) def test_fs_node_is_xx(nc): nc.files.delete("test_root_folder", not_fail=True)