[Bitbucket] Added Download class to retrieve pipeline-generated artifacts (Cloud)

Author: Denis Abakumov

atlassian/bitbucket/cloud/common/users.py

@@ -1,3 +1,5 @@
+from datetime import datetime
+
 from ..base import BitbucketCloudBase
 
 
@@ -31,6 +33,49 @@ def avatar(self):
         return self.get_data("links")["avatar"]["href"]
 
 
+class AppUser(BitbucketCloudBase):
+    def __init__(self, url, data, *args, **kwargs):
+        super(AppUser, self).__init__(url, *args, data=data, expected_type="app_user", **kwargs)
+
+    @property
+    def display_name(self):
+        """User display name"""
+        return str(self.get_data("display_name"))
+
+    @property
+    def account_id(self):
+        """User account id"""
+        return self.get_data("account_id")
+
+    @property
+    def account_status(self):
+        """User account status"""
+        return self.get_data("account_status")
+
+    @property
+    def created_on(self):
+        """User creation date"""
+        created_on = self.get_data("created_on")
+        if created_on is None:
+            return None
+        return datetime.strptime(created_on, self.CONF_TIMEFORMAT)
+
+    @property
+    def uuid(self):
+        """User id"""
+        return self.get_data("uuid")
+
+    @property
+    def kind(self):
+        """User kind"""
+        return self.get_data("kind")
+
+    @property
+    def links(self):
+        """User links"""
+        return self.get_data("links")
+
+
 class Participant(BitbucketCloudBase):
     ROLE_REVIEWER = "REVIEWER"
     ROLE_PARTICIPANT = "PARTICIPANT"

atlassian/bitbucket/cloud/repositories/__init__.py

@@ -1,18 +1,20 @@
 # coding=utf-8
 
 from requests import HTTPError
-from ..base import BitbucketCloudBase
-from .issues import Issues
+
 from .branchRestrictions import BranchRestrictions
 from .commits import Commits
-from .hooks import Hooks
 from .defaultReviewers import DefaultReviewers
 from .deploymentEnvironments import DeploymentEnvironments
+from .downloads import Downloads
 from .groupPermissions import GroupPermissions
+from .hooks import Hooks
+from .issues import Issues
 from .pipelines import Pipelines
 from .pullRequests import PullRequests
 from .refs import Branches, Tags
 from .repositoryVariables import RepositoryVariables
+from ..base import BitbucketCloudBase
 
 
 class RepositoriesBase(BitbucketCloudBase):
@@ -277,6 +279,7 @@ def __init__(self, data, *args, **kwargs):
             f"{self.url}/pipelines_config/variables", **self._new_session_args
         )
         self.__tags = Tags(f"{self.url}/refs/tags", **self._new_session_args)
+        self.__downloads = Downloads(f"{self.url}/downloads", **self._new_session_args)
 
     def update(self, **kwargs):
         """
@@ -433,3 +436,8 @@ def repository_variables(self):
     def tags(self):
         """The repository tags."""
         return self.__tags
+
+    @property
+    def downloads(self):
+        """The repository downloads"""
+        return self.__downloads

atlassian/bitbucket/cloud/repositories/downloads.py

@@ -0,0 +1,75 @@
+from datetime import datetime
+
+from ..base import BitbucketCloudBase
+from ..common.users import AppUser
+
+
+class Download(BitbucketCloudBase):
+    def __init__(self, url, data, *args, **kwargs):
+        super(Download, self).__init__(url, *args, data=data, expected_type="download", **kwargs)
+
+    @property
+    def name(self):
+        """The download file name"""
+        return self.get_data("name")
+
+    @property
+    def size(self):
+        """The download size in bytes"""
+        return self.get_data("size")
+
+    @property
+    def created_on(self):
+        """The download created date"""
+        created_on = self.get_data("created_on")
+        if created_on is None:
+            return None
+        return datetime.strptime(created_on, self.CONF_TIMEFORMAT)
+
+    @property
+    def user(self):
+        """The user who created this download"""
+        return AppUser(None, self.get_data("user"))
+
+    @property
+    def downloads(self):
+        """The number of times this file was downloaded"""
+        return self.get_data("downloads")
+
+    @property
+    def links(self):
+        """The download links"""
+        return self.get_data("links")
+
+    def get_content(self):
+        """Download file content"""
+        return self.get(self.url, absolute=True)
+
+
+class Downloads(BitbucketCloudBase):
+    def __init__(self, url, *args, **kwargs):
+        super(Downloads, self).__init__(url, *args, **kwargs)
+
+    def __get_object(self, data):
+        url = self.url_joiner(self.url, data["name"])
+        if data["links"] and data["links"]["self"] and data["links"]["self"]["href"]:
+            url = data["links"]["self"]["href"]
+
+        return Download(
+            url,
+            data,
+            **self._new_session_args
+        )  # fmt: skip
+
+    def each(self):
+        """
+        Returns the list of downloads in this repository.
+
+        :return: A generator for the Download objects
+
+        API docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-downloads/#api-repositories-workspace-repo-slug-downloads-get
+        """
+        for issue in self._get_paged(None):
+            yield self.__get_object(issue)
+
+        return