FileAPI: download_directory_as_zip (#73)

Signed-off-by: Alexander Piskun <[email protected]>

Author: bigcat88

.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 &

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
 

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)

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.
 

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)