Skip to content

Artifact API

This page describes the ArtifactAPI class.

ArtifactAPI

Class that allows for interacting with artifacts via API.

Attributes:

Name Type Description
client object

client for sending requests to Typhoon Test Hub API
Source code in tth\api\artifact.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class ArtifactAPI:
    """
    Class that allows for interacting with artifacts via API.

    Attributes:
        client (object): client for sending requests to Typhoon Test Hub API
    """

    def __init__(self, client):
        self.client = client
        self.__locks = {}
        self.__percentages = {}

    def __create_upload_callback(self, execution, name, no_progress_output):
        def callback(monitor):
            if not no_progress_output:
                encoder_len = monitor.encoder.len

                percentage = math.floor(monitor.bytes_read / encoder_len * 100)
                if str(execution) in self.__locks:
                    with self.__locks[str(execution)]:
                        if percentage > self.__percentages[str(execution)]:
                            self.__percentages[str(execution)] = percentage
                            progress_message = utils.generate_progress_string(monitor.bytes_read, encoder_len)
                            log.info(f"Uploading artifact: {name} {progress_message}")

        return callback

    def upload_artifact(self, artifact_path, zip_dir=False, no_progress_output=True, execution_id=None,
                        retention_until=None):
        """
        Upload a new artifact.

        Args:
            artifact_path (str): Path of artifact
            zip_dir (bool, optional): If True, artifact_path should point to a zipped directory which should be
                uploaded as an artifact; artifact_path should point to a file which should be uploaded as an artifact;
                default value is False
            no_progress_output (bool, optional): If True, no upload progress logs are shown; default value is True
            execution_id (int, optional): Identifier of the Execution to which the artifact should be assigned
            retention_until (datetime, optional): Timestamp until artifact should be preserved

        Returns:
            identifier (int): Identifier of created artifact

        Raises:
            APIException: Response status code not 201
            InvalidPathException: Invalid artifact path provided
        """
        artifact_absolute_path = os.path.abspath(artifact_path)
        if not os.path.exists(artifact_absolute_path):
            raise InvalidPathException(f"unknown path {artifact_path}")
        if zip_dir:
            if not os.path.isdir(artifact_absolute_path):
                raise InvalidPathException(f"{artifact_path} should be directory")
        else:
            if not os.path.isfile(artifact_absolute_path):
                raise InvalidPathException(f"{artifact_path} should be file because zip_dir is True")

        head, tail = os.path.split(artifact_absolute_path)
        artifact_info = {"name": tail, "path": artifact_absolute_path}
        temp_dir = None
        if zip_dir:
            temp_dir = tempfile.mkdtemp()
            zip_path = report_handler.create_zip_archive_in_directory(tail, artifact_absolute_path, temp_dir)
            head, tail = os.path.split(zip_path)
            artifact_info = {"name": tail, "path": zip_path}
        retention = retention_until
        if retention_until is not None:
            if not isinstance(retention_until, datetime):
                raise InvalidDataException("retention until timestamp should be datetime instance")

            retention = retention_until.astimezone(timezone.utc)
            if retention < datetime.now(timezone.utc):
                raise InvalidDataException("retention until timestamp cannot be in the past")
            retention = retention.isoformat()

        encoder = MultipartEncoder({
            "file": (artifact_info["name"], open(artifact_info["path"], "rb"), "text/plain"),
            "execution_id": str(execution_id),
            "retention_until": retention
        })

        upload_identifier = str(uuid.uuid4())
        callback = self.__create_upload_callback(upload_identifier, artifact_info["name"], no_progress_output)
        monitor = MultipartEncoderMonitor(encoder, callback)

        if not no_progress_output:
            log.info(f"Uploading artifact: {artifact_info['name']} 0%")

        self.__locks[upload_identifier] = threading.Lock()
        self.__percentages[upload_identifier] = 0
        headers = {**{"Content-Type": monitor.content_type}, **self.client.get_request_headers()}
        response = requests.post(f"{self.client.url}/api/artifacts/upload/manual", data=monitor, headers=headers)
        del self.__locks[upload_identifier]
        del self.__percentages[upload_identifier]

        if zip_dir and temp_dir:
            shutil.rmtree(temp_dir, ignore_errors=True)

        if response.status_code == 201:
            artifact_id = response.json()
            return int(artifact_id)
        else:
            raise APIException(response)

    def download_artifact(self, artifact_id, file_name, destination='', no_progress_output=True):
        """
        Download an existing artifact from Typhoon Test Hub.

        Args:
            artifact_id (int): Identifier of artifact
            file_name (str): Name of file after download
            destination (str): Directory where file is downloaded; default value is an empty string (file downloaded in
                working directory)
            no_progress_output (bool, optional): If True, no download progress logs are shown; default value is True

        Returns:
            path (str): Absolute path of downloaded file

        Raises:
            APIException: Response status code not 200
        """
        artifact_url = f"{self.client.url}/api/downloads/artifacts/{str(artifact_id)}"
        headers = {**{"User-Agent": "Typhoon Officer"}, **self.client.get_request_headers()}
        if not utils.is_artifact_downloadable(artifact_url, headers):
            raise APIException("Artifact is not available")
        download_identifier = str(uuid.uuid4())
        file_path = os.path.abspath(os.path.join(destination, file_name))
        self.__locks[download_identifier] = threading.Lock()
        self.__percentages[download_identifier] = 0

        if not no_progress_output:
            log.info(f"Downloading artifact: {file_name} {utils.generate_progress_string(0, 0)}")

        with requests.post(artifact_url, allow_redirects=True, headers=headers) as response:
            if response.status_code != 200:
                raise APIException(response)

            total_size = int(response.headers.get("Content-Length", 0))
            chunk_size = 10 ** 6
            with open(file_path, mode="wb") as file:
                size = 0
                for data in response.iter_content(chunk_size=chunk_size):
                    size += file.write(data)
                    if not no_progress_output:
                        if total_size > 0:
                            percentage = math.floor(size / total_size * 100)
                            if download_identifier in self.__locks:
                                with self.__locks[download_identifier]:
                                    if percentage > self.__percentages[download_identifier]:
                                        self.__percentages[download_identifier] = percentage
                                        log.info(f"Downloading artifact: {file_name} "
                                                 f"{utils.generate_progress_string(size, total_size)}")
        del self.__locks[download_identifier]
        del self.__percentages[download_identifier]
        return file_path

download_artifact(artifact_id, file_name, destination='', no_progress_output=True)

Download an existing artifact from Typhoon Test Hub.

Parameters:

Name Type Description Default
artifact_id int

Identifier of artifact

 required
file_name str

Name of file after download

 required
destination str

Directory where file is downloaded; default value is an empty string (file downloaded in working directory)

 ''
no_progress_output bool

If True, no download progress logs are shown; default value is True

 True

Returns:

Name Type Description
path str

Absolute path of downloaded file

Raises:

Type Description
APIException

Response status code not 200
Source code in tth\api\artifact.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def download_artifact(self, artifact_id, file_name, destination='', no_progress_output=True):
    """
    Download an existing artifact from Typhoon Test Hub.

    Args:
        artifact_id (int): Identifier of artifact
        file_name (str): Name of file after download
        destination (str): Directory where file is downloaded; default value is an empty string (file downloaded in
            working directory)
        no_progress_output (bool, optional): If True, no download progress logs are shown; default value is True

    Returns:
        path (str): Absolute path of downloaded file

    Raises:
        APIException: Response status code not 200
    """
    artifact_url = f"{self.client.url}/api/downloads/artifacts/{str(artifact_id)}"
    headers = {**{"User-Agent": "Typhoon Officer"}, **self.client.get_request_headers()}
    if not utils.is_artifact_downloadable(artifact_url, headers):
        raise APIException("Artifact is not available")
    download_identifier = str(uuid.uuid4())
    file_path = os.path.abspath(os.path.join(destination, file_name))
    self.__locks[download_identifier] = threading.Lock()
    self.__percentages[download_identifier] = 0

    if not no_progress_output:
        log.info(f"Downloading artifact: {file_name} {utils.generate_progress_string(0, 0)}")

    with requests.post(artifact_url, allow_redirects=True, headers=headers) as response:
        if response.status_code != 200:
            raise APIException(response)

        total_size = int(response.headers.get("Content-Length", 0))
        chunk_size = 10 ** 6
        with open(file_path, mode="wb") as file:
            size = 0
            for data in response.iter_content(chunk_size=chunk_size):
                size += file.write(data)
                if not no_progress_output:
                    if total_size > 0:
                        percentage = math.floor(size / total_size * 100)
                        if download_identifier in self.__locks:
                            with self.__locks[download_identifier]:
                                if percentage > self.__percentages[download_identifier]:
                                    self.__percentages[download_identifier] = percentage
                                    log.info(f"Downloading artifact: {file_name} "
                                             f"{utils.generate_progress_string(size, total_size)}")
    del self.__locks[download_identifier]
    del self.__percentages[download_identifier]
    return file_path

upload_artifact(artifact_path, zip_dir=False, no_progress_output=True, execution_id=None, retention_until=None)

Upload a new artifact.

Parameters:

Name Type Description Default
artifact_path str

Path of artifact

 required
zip_dir bool

If True, artifact_path should point to a zipped directory which should be uploaded as an artifact; artifact_path should point to a file which should be uploaded as an artifact; default value is False

 False
no_progress_output bool

If True, no upload progress logs are shown; default value is True

 True
execution_id int

Identifier of the Execution to which the artifact should be assigned

 None
retention_until datetime

Timestamp until artifact should be preserved

 None

Returns:

Name Type Description
identifier int

Identifier of created artifact

Raises:

Type Description
APIException

Response status code not 201
InvalidPathException

Invalid artifact path provided
Source code in tth\api\artifact.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def upload_artifact(self, artifact_path, zip_dir=False, no_progress_output=True, execution_id=None,
                    retention_until=None):
    """
    Upload a new artifact.

    Args:
        artifact_path (str): Path of artifact
        zip_dir (bool, optional): If True, artifact_path should point to a zipped directory which should be
            uploaded as an artifact; artifact_path should point to a file which should be uploaded as an artifact;
            default value is False
        no_progress_output (bool, optional): If True, no upload progress logs are shown; default value is True
        execution_id (int, optional): Identifier of the Execution to which the artifact should be assigned
        retention_until (datetime, optional): Timestamp until artifact should be preserved

    Returns:
        identifier (int): Identifier of created artifact

    Raises:
        APIException: Response status code not 201
        InvalidPathException: Invalid artifact path provided
    """
    artifact_absolute_path = os.path.abspath(artifact_path)
    if not os.path.exists(artifact_absolute_path):
        raise InvalidPathException(f"unknown path {artifact_path}")
    if zip_dir:
        if not os.path.isdir(artifact_absolute_path):
            raise InvalidPathException(f"{artifact_path} should be directory")
    else:
        if not os.path.isfile(artifact_absolute_path):
            raise InvalidPathException(f"{artifact_path} should be file because zip_dir is True")

    head, tail = os.path.split(artifact_absolute_path)
    artifact_info = {"name": tail, "path": artifact_absolute_path}
    temp_dir = None
    if zip_dir:
        temp_dir = tempfile.mkdtemp()
        zip_path = report_handler.create_zip_archive_in_directory(tail, artifact_absolute_path, temp_dir)
        head, tail = os.path.split(zip_path)
        artifact_info = {"name": tail, "path": zip_path}
    retention = retention_until
    if retention_until is not None:
        if not isinstance(retention_until, datetime):
            raise InvalidDataException("retention until timestamp should be datetime instance")

        retention = retention_until.astimezone(timezone.utc)
        if retention < datetime.now(timezone.utc):
            raise InvalidDataException("retention until timestamp cannot be in the past")
        retention = retention.isoformat()

    encoder = MultipartEncoder({
        "file": (artifact_info["name"], open(artifact_info["path"], "rb"), "text/plain"),
        "execution_id": str(execution_id),
        "retention_until": retention
    })

    upload_identifier = str(uuid.uuid4())
    callback = self.__create_upload_callback(upload_identifier, artifact_info["name"], no_progress_output)
    monitor = MultipartEncoderMonitor(encoder, callback)

    if not no_progress_output:
        log.info(f"Uploading artifact: {artifact_info['name']} 0%")

    self.__locks[upload_identifier] = threading.Lock()
    self.__percentages[upload_identifier] = 0
    headers = {**{"Content-Type": monitor.content_type}, **self.client.get_request_headers()}
    response = requests.post(f"{self.client.url}/api/artifacts/upload/manual", data=monitor, headers=headers)
    del self.__locks[upload_identifier]
    del self.__percentages[upload_identifier]

    if zip_dir and temp_dir:
        shutil.rmtree(temp_dir, ignore_errors=True)

    if response.status_code == 201:
        artifact_id = response.json()
        return int(artifact_id)
    else:
        raise APIException(response)