Skip to content

Execution API

This page describes the ExecutionAPI class.

ExecutionAPI

Class that allows for interacting with executions via API.

Attributes:

Name Type Description
client object

client for sending requests to Typhoon Test Hub API
Source code in tth\api\execution.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 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
class ExecutionAPI:
    """
    Class that allows for interacting with executions via API.

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

    def __init__(self, client):
        self.client = client

    def get_info(self, execution_id):
        """
        Obtain details of the Execution with the provided execution identifier.

        Args:
            execution_id (int): Identifier of execution

        Returns:
            data (dict): Execution information

        Raises:
            APIException: Response status code not 200
        """
        response = requests.get(f"{self.client.url}/api/executions/status/{str(execution_id)}",
                                headers=self.client.get_request_headers())
        if response.status_code == 200:
            return response.json()
        else:
            raise APIException(response)

    def get_status(self, execution_id):
        """
        Obtain status of the Execution with the provided execution identifier.

        Args:
            execution_id (int): Identifier of execution

        Returns:
            status (str): Execution status

        Raises:
            APIException: Response status code not 200
        """
        response = requests.get(f"{self.client.url}/api/executions/status/{str(execution_id)}",
                                headers=self.client.get_request_headers())
        if response.status_code == 200:
            return response.json()['status']
        else:
            raise APIException(response)

    def wait_until_finished(self, execution_id, interval=10, timeout=None, ignore_errors=False):
        """
        Periodically check the selected Execution's status and waits for the Execution to finish.

        Args:
            execution_id (int): Identifier of execution
            interval (int): Short polling interval for checking execution status
            timeout (int): Timeout (in seconds) of waiting for execution to finish
            ignore_errors (bool): Ignore APIException instances if happen (useful when internet connection is unstable)

        Returns:
            status(str): Execution status

        Raises:
            APIException: Response status code not 200 if ignore_error is False
        """
        start_epoch = int(time.time())
        while True:
            try:
                status = self.get_status(execution_id)
                if status not in ['QUEUED', 'RUNNING']:
                    break

                if timeout is not None:
                    current_epoch = int(time.time())
                    if (current_epoch - start_epoch) > timeout:
                        raise TimeoutError(f'Timeout of {str(timeout)} seconds reached.')
            except APIException as e:
                if not ignore_errors:
                    raise e

            time.sleep(int(interval))
        return status

    def download_artifacts(self, execution_id, expression, destination='', no_progress_output=True):
        """
        Download artifacts belonging to the Execution matching provided expression.

        Args:
            execution_id (int): Identifier of execution
            expression (str): Artifacts selection expression
            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

        Raises:
            APIException: Response status code not 200
        """
        request_data = {'option': 'SPECIFIC', 'execution': str(execution_id),
                        'chosen_execution': str(execution_id), 'collect': expression}
        response = requests.post(f"{self.client.url}/api/downloads/available/artifacts",
                                 headers=self.client.get_request_headers(), json=request_data)
        if response.status_code == 200:
            artifacts = response.json()
            if artifacts is None or len(artifacts) == 0:
                raise APIException("No artifacts matching expression were found.")

            for artifact in artifacts:
                self.client.artifact.download_artifact(artifact["id"], artifact["name"],
                                                       destination, no_progress_output)
        else:
            raise APIException(response)

download_artifacts(execution_id, expression, destination='', no_progress_output=True)

Download artifacts belonging to the Execution matching provided expression.

Parameters:

Name Type Description Default
execution_id int

Identifier of execution

 required
expression str

Artifacts selection expression

 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

Raises:

Type Description
APIException

Response status code not 200
Source code in tth\api\execution.py 
 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
def download_artifacts(self, execution_id, expression, destination='', no_progress_output=True):
    """
    Download artifacts belonging to the Execution matching provided expression.

    Args:
        execution_id (int): Identifier of execution
        expression (str): Artifacts selection expression
        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

    Raises:
        APIException: Response status code not 200
    """
    request_data = {'option': 'SPECIFIC', 'execution': str(execution_id),
                    'chosen_execution': str(execution_id), 'collect': expression}
    response = requests.post(f"{self.client.url}/api/downloads/available/artifacts",
                             headers=self.client.get_request_headers(), json=request_data)
    if response.status_code == 200:
        artifacts = response.json()
        if artifacts is None or len(artifacts) == 0:
            raise APIException("No artifacts matching expression were found.")

        for artifact in artifacts:
            self.client.artifact.download_artifact(artifact["id"], artifact["name"],
                                                   destination, no_progress_output)
    else:
        raise APIException(response)

get_info(execution_id)

Obtain details of the Execution with the provided execution identifier.

Parameters:

Name Type Description Default
execution_id int

Identifier of execution

 required

Returns:

Name Type Description
data dict

Execution information

Raises:

Type Description
APIException

Response status code not 200
Source code in tth\api\execution.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def get_info(self, execution_id):
    """
    Obtain details of the Execution with the provided execution identifier.

    Args:
        execution_id (int): Identifier of execution

    Returns:
        data (dict): Execution information

    Raises:
        APIException: Response status code not 200
    """
    response = requests.get(f"{self.client.url}/api/executions/status/{str(execution_id)}",
                            headers=self.client.get_request_headers())
    if response.status_code == 200:
        return response.json()
    else:
        raise APIException(response)

get_status(execution_id)

Obtain status of the Execution with the provided execution identifier.

Parameters:

Name Type Description Default
execution_id int

Identifier of execution

 required

Returns:

Name Type Description
status str

Execution status

Raises:

Type Description
APIException

Response status code not 200
Source code in tth\api\execution.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def get_status(self, execution_id):
    """
    Obtain status of the Execution with the provided execution identifier.

    Args:
        execution_id (int): Identifier of execution

    Returns:
        status (str): Execution status

    Raises:
        APIException: Response status code not 200
    """
    response = requests.get(f"{self.client.url}/api/executions/status/{str(execution_id)}",
                            headers=self.client.get_request_headers())
    if response.status_code == 200:
        return response.json()['status']
    else:
        raise APIException(response)

wait_until_finished(execution_id, interval=10, timeout=None, ignore_errors=False)

Periodically check the selected Execution's status and waits for the Execution to finish.

Parameters:

Name Type Description Default
execution_id int

Identifier of execution

 required
interval int

Short polling interval for checking execution status

 10
timeout int

Timeout (in seconds) of waiting for execution to finish

 None
ignore_errors bool

Ignore APIException instances if happen (useful when internet connection is unstable)

 False

Returns:

Name Type Description
status str

Execution status

Raises:

Type Description
APIException

Response status code not 200 if ignore_error is False
Source code in tth\api\execution.py 
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
def wait_until_finished(self, execution_id, interval=10, timeout=None, ignore_errors=False):
    """
    Periodically check the selected Execution's status and waits for the Execution to finish.

    Args:
        execution_id (int): Identifier of execution
        interval (int): Short polling interval for checking execution status
        timeout (int): Timeout (in seconds) of waiting for execution to finish
        ignore_errors (bool): Ignore APIException instances if happen (useful when internet connection is unstable)

    Returns:
        status(str): Execution status

    Raises:
        APIException: Response status code not 200 if ignore_error is False
    """
    start_epoch = int(time.time())
    while True:
        try:
            status = self.get_status(execution_id)
            if status not in ['QUEUED', 'RUNNING']:
                break

            if timeout is not None:
                current_epoch = int(time.time())
                if (current_epoch - start_epoch) > timeout:
                    raise TimeoutError(f'Timeout of {str(timeout)} seconds reached.')
        except APIException as e:
            if not ignore_errors:
                raise e

        time.sleep(int(interval))
    return status