Merge pull request #64 from stefanseefeld/recursive

Add support for recursive up- and downloads

pyartifactory/objects.py

@@ -3,7 +3,9 @@
 """
 import warnings
 import logging
-from typing import List, Optional, Dict, Tuple, Union
+import os
+from os.path import isdir, join
+from typing import List, Optional, Dict, Tuple, Union, Iterator
 
 from pathlib import Path
 import requests
@@ -752,6 +754,27 @@ def delete(self, permission_name: str) -> None:
 class ArtifactoryArtifact(ArtifactoryObject):
     """Models an artifactory artifact."""
 
+    def _walk(
+        self, artifact_path: str, topdown: bool = True
+    ) -> Iterator[ArtifactInfoResponse]:
+        """Iterate over artifact (file or directory) recursively.
+
+        :param artifact_path: Path to file or folder in Artifactory
+        :param topdown: True for a top-down (directory first) traversal
+        """
+        info = self.info(artifact_path)
+        if not isinstance(info, ArtifactFolderInfoResponse):
+            yield info
+        else:
+            if topdown:
+                yield info
+            for subdir in (child for child in info.children if child.folder is True):
+                yield from self._walk(artifact_path + subdir.uri, topdown=topdown)
+            for file in (child for child in info.children if child.folder is not True):
+                yield from self._walk(artifact_path + file.uri, topdown=topdown)
+            if not topdown:
+                yield info
+
     def info(self, artifact_path: str) -> ArtifactInfoResponse:
         """
         Retrieve information about a file or a folder
@@ -781,20 +804,30 @@ def deploy(
         self, local_file_location: str, artifact_path: str
     ) -> ArtifactInfoResponse:
         """
-        :param artifact_path: Path to file in Artifactory
-        :param local_file_location: Location of the file to deploy
-        """
-        artifact_path = artifact_path.lstrip("/")
-        local_filename = artifact_path.split("/")[-1]
-        with open(local_file_location, "rb") as file:
-            self._put(f"{artifact_path}", data=file)
-            logger.debug("Artifact %s successfully deployed", local_filename)
-            return self.info(artifact_path)
-
-    def download(self, artifact_path: str, local_directory_path: str = None) -> str:
+        Deploy a file or directory.
+        :param artifact_path: Path to artifactory in Artifactory
+        :param local_file_location: Location of the file or folder to deploy
         """
+        if isdir(local_file_location):
+            for root, _, files in os.walk(local_file_location):
+                new_root = f"{artifact_path}/{root[len(local_file_location):]}"
+                for file in files:
+                    self.deploy(join(root, file), f"{new_root}/{file}")
+        else:
+            artifact_path = artifact_path.lstrip("/")
+            local_filename = artifact_path.split("/")[-1]
+            # we need to silence a bogus mypy error as `file` is already used above
+            # with a different type. See https://github.com/python/mypy/issues/6232
+            with open(local_file_location, "rb") as file:  # type: ignore[assignment]
+                self._put(f"{artifact_path}", data=file)
+                logger.debug("Artifact %s successfully deployed", local_filename)
+        return self.info(artifact_path)
+
+    def _download(self, artifact_path: str, local_directory_path: str = None) -> str:
+        """
+        Download artifact (file) into local directory.
         :param artifact_path: Path to file in Artifactory
-        :param local_directory_path: Local path to where the artifact will be imported
+        :param local_directory_path: Local path to where the artifact will be downloaded
         :return: File name
         """
         artifact_path = artifact_path.lstrip("/")
@@ -814,6 +847,25 @@ def download(self, artifact_path: str, local_directory_path: str = None) -> str:
         logger.debug("Artifact %s successfully downloaded", local_filename)
         return local_file_full_path
 
+    def download(self, artifact_path: str, local_directory_path: str = ".") -> str:
+        """
+        Download artifact (file or directory) into local directory.
+        :param artifact_path: Path to file or directory in Artifactory
+        :param local_directory_path: Local path to where the artifact will be downloaded
+        :return: File name
+        """
+        artifact_path = artifact_path.rstrip("/")
+        basename = artifact_path.split("/")[-1]
+        prefix = artifact_path.rsplit("/", 1)[0] + "/" if "/" in artifact_path else ""
+        for art in self._walk(artifact_path):
+            full_path = art.repo + art.path
+            local_path = join(local_directory_path, full_path[len(prefix) :])
+            if isinstance(art, ArtifactFolderInfoResponse):
+                os.makedirs(local_path, exist_ok=True)
+            else:
+                self._download(art.repo + art.path, local_path.rsplit("/", 1)[0])
+        return f"{local_directory_path}/{basename}"
+
     def properties(
         self, artifact_path: str, properties: Optional[List[str]] = None
     ) -> ArtifactPropertiesResponse:

tests/test_artifacts.py

@@ -15,8 +15,8 @@
 
 URL = "http://localhost:8080/artifactory"
 AUTH = ("user", "password_or_apiKey")
-ARTIFACT_FOLDER = "my_repository"
-ARTIFACT_PATH = f"{ARTIFACT_FOLDER}/file.txt"
+ARTIFACT_REPO = "my_repository"
+ARTIFACT_PATH = f"{ARTIFACT_REPO}/file.txt"
 ARTIFACT_NEW_PATH = "my-second-repository/file.txt"
 ARTIFACT_SHORT_PATH = "/file.txt"
 LOCAL_FILE_LOCATION = "tests/test_artifacts.py"
@@ -28,8 +28,8 @@
     properties={"prop1": ["value"], "prop2": ["another value", "with multiple parts"]},
 )
 FOLDER_INFO_RESPONSE = {
-    "uri": f"{URL}/api/storage/{ARTIFACT_FOLDER}",
-    "repo": ARTIFACT_FOLDER,
+    "uri": f"{URL}/api/storage/{ARTIFACT_REPO}",
+    "repo": ARTIFACT_REPO,
     "path": "/",
     "created": "2019-06-06T13:19:14.514Z",
     "createdBy": "userY",
@@ -42,9 +42,19 @@
     ],
 }
 FOLDER_INFO = ArtifactFolderInfoResponse(**FOLDER_INFO_RESPONSE)
+
+CHILD1_FOLDER_INFO_RESPONSE = FOLDER_INFO_RESPONSE.copy()
+CHILD1_FOLDER_INFO_RESPONSE["uri"] = f"{URL}/api/storage/{ARTIFACT_REPO}/child1"
+CHILD1_FOLDER_INFO_RESPONSE["path"] = "/child1"
+CHILD1_FOLDER_INFO_RESPONSE["children"] = [
+    {"uri": "/grandchild", "folder": "false"},
+]
+CHILD1_FOLDER_INFO = ArtifactFolderInfoResponse(**CHILD1_FOLDER_INFO_RESPONSE)
+
+
 FILE_INFO_RESPONSE = {
-    "repo": ARTIFACT_FOLDER,
-    "path": ARTIFACT_PATH,
+    "repo": ARTIFACT_REPO,
+    "path": ARTIFACT_SHORT_PATH,
     "created": "2019-06-06T13:19:14.514Z",
     "createdBy": "userY",
     "lastModified": "2019-06-06T13:19:14.514Z",
@@ -65,6 +75,17 @@
 }
 FILE_INFO = ArtifactFileInfoResponse(**FILE_INFO_RESPONSE)
 
+CHILD2_INFO_RESPONSE = FILE_INFO_RESPONSE.copy()
+CHILD2_INFO_RESPONSE["uri"] = f"{URL}/api/storage/{ARTIFACT_REPO}/child2"
+CHILD2_INFO_RESPONSE["path"] = "/child2"
+CHILD2_FILE_INFO = ArtifactFileInfoResponse(**CHILD2_INFO_RESPONSE)
+
+GRANDCHILD_INFO_RESPONSE = FILE_INFO_RESPONSE.copy()
+GRANDCHILD_INFO_RESPONSE["uri"] = f"{URL}/api/storage/{ARTIFACT_REPO}/child1/grandchild"
+GRANDCHILD_INFO_RESPONSE["path"] = "/child1/grandchild"
+GRANDCHILD_FILE_INFO = ArtifactFileInfoResponse(**GRANDCHILD_INFO_RESPONSE)
+
+
 ARTIFACT_STATS = ArtifactStatsResponse(
     uri="my_uri",
     downloadCount=0,
@@ -78,12 +99,12 @@
 def test_get_artifact_folder_info_success():
     responses.add(
         responses.GET,
-        f"{URL}/api/storage/{ARTIFACT_FOLDER}",
+        f"{URL}/api/storage/{ARTIFACT_REPO}",
         status=200,
         json=FOLDER_INFO_RESPONSE,
     )
     artifactory = ArtifactoryArtifact(AuthModel(url=URL, auth=AUTH))
-    artifact = artifactory.info(ARTIFACT_FOLDER)
+    artifact = artifactory.info(ARTIFACT_REPO)
     assert isinstance(artifact, ArtifactFolderInfoResponse)
     assert artifact.dict() == FOLDER_INFO.dict()
 
@@ -122,6 +143,12 @@ def test_deploy_artifact_success(mocker):
 @responses.activate
 def test_download_artifact_success(tmp_path):
     artifact_name = ARTIFACT_PATH.split("/")[1]
+    responses.add(
+        responses.GET,
+        f"{URL}/api/storage/{ARTIFACT_PATH}",
+        status=200,
+        json=FILE_INFO_RESPONSE,
+    )
     responses.add(
         responses.GET, f"{URL}/{ARTIFACT_PATH}", json=artifact_name, status=200
     )
@@ -133,6 +160,52 @@ def test_download_artifact_success(tmp_path):
     assert (tmp_path / artifact_name).is_file()
 
 
+@responses.activate
+def test_download_folder_success(tmp_path):
+    # artifact_name = ARTIFACT_PATH.split("/")[1]
+    responses.add(
+        responses.GET,
+        f"{URL}/api/storage/{ARTIFACT_REPO}",
+        status=200,
+        json=FOLDER_INFO_RESPONSE,
+    )
+    responses.add(
+        responses.GET,
+        f"{URL}/api/storage/{ARTIFACT_REPO}/child1",
+        status=200,
+        json=CHILD1_FOLDER_INFO_RESPONSE,
+    )
+    responses.add(
+        responses.GET,
+        f"{URL}/api/storage/{ARTIFACT_REPO}/child2",
+        status=200,
+        json=CHILD2_INFO_RESPONSE,
+    )
+    responses.add(
+        responses.GET,
+        f"{URL}/api/storage/{ARTIFACT_REPO}/child1/grandchild",
+        status=200,
+        json=GRANDCHILD_INFO_RESPONSE,
+    )
+    responses.add(responses.GET, f"{URL}/{ARTIFACT_REPO}", json="/", status=200)
+    responses.add(
+        responses.GET,
+        f"{URL}/{ARTIFACT_REPO}/child1/grandchild",
+        json="/child1/grandchild",
+        status=200,
+    )
+    responses.add(
+        responses.GET, f"{URL}/{ARTIFACT_REPO}/child2", json="/child2", status=200
+    )
+
+    artifactory = ArtifactoryArtifact(AuthModel(url=URL, auth=AUTH))
+    artifact = artifactory.download(f"{ARTIFACT_REPO}/", str(tmp_path.resolve()))
+
+    assert artifact == f"{tmp_path.resolve()}/{ARTIFACT_REPO}"
+    assert (tmp_path / f"{ARTIFACT_REPO}" / "child1" / "grandchild").is_file()
+    assert (tmp_path / f"{ARTIFACT_REPO}" / "child2").is_file()
+
+
 @responses.activate
 def test_get_artifact_single_property_success():
     responses.add(