Skip to content

BatchJobResult

Batch job result.

Source code in hume/_batch/batch_job_result.py 
 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
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
177
178
179
180
181
182
183
class BatchJobResult:
    """Batch job result."""

    def __init__(
        self,
        *,
        configs: Dict[ModelType, ModelConfigBase],
        urls: List[str],
        status: BatchJobStatus,
        predictions_url: Optional[str] = None,
        artifacts_url: Optional[str] = None,
        errors_url: Optional[str] = None,
        error_message: Optional[str] = None,
        job_start_time: Optional[int] = None,
        job_end_time: Optional[int] = None,
    ):
        """Construct a BatchJobResult.

        Args:
            configs (Dict[ModelType, ModelConfigBase]): Configurations for the `BatchJob`.
            urls (List[str]): URLs processed in the `BatchJob`.
            status (BatchJobStatus): Status of `BatchJob`.
            predictions_url (Optional[str]): URL to predictions file.
            artifacts_url (Optional[str]): URL to artifacts zip archive.
            errors_url (Optional[str]): URL to errors file.
            error_message (Optional[str]): Error message for request.
            job_start_time (Optional[int]): Time when job started.
            job_end_time (Optional[int]): Time when job completed.
        """
        self.configs = configs
        self.urls = urls
        self.status = status
        self.predictions_url = predictions_url
        self.artifacts_url = artifacts_url
        self.errors_url = errors_url
        self.error_message = error_message
        self.job_start_time = job_start_time
        self.job_end_time = job_end_time

    def download_predictions(self, filepath: Optional[Union[str, Path]] = None) -> None:
        """Download `BatchJob` predictions file.

        Args:
            filepath (Optional[Union[str, Path]]): Filepath where predictions will be downloaded.
        """
        if self.predictions_url is None:
            raise HumeClientException("Could not download job predictions. No predictions found on job result.")
        urlretrieve(self.predictions_url, filepath)

    def download_artifacts(self, filepath: Optional[Union[str, Path]] = None) -> None:
        """Download `BatchJob` artifacts zip archive.

        Args:
            filepath (Optional[Union[str, Path]]): Filepath where artifacts zip archive will be downloaded.
        """
        if self.artifacts_url is None:
            raise HumeClientException("Could not download job artifacts. No artifacts found on job result.")
        urlretrieve(self.artifacts_url, filepath)

    def download_errors(self, filepath: Optional[Union[str, Path]] = None) -> None:
        """Download `BatchJob` errors file.

        Args:
            filepath (Optional[Union[str, Path]]): Filepath where errors will be downloaded.
        """
        if self.errors_url is None:
            raise HumeClientException("Could not download job errors. No errors found on job result.")
        urlretrieve(self.errors_url, filepath)

    def get_error_message(self) -> Optional[str]:
        """Get any available error messages on the job.

        Returns:
            Optional[str]: A string with the error message if there was an error, otherwise `None`.
        """
        return self.error_message

    def get_run_time(self) -> Optional[int]:
        """Get the total time in seconds it took for the job to run if the job is in a terminal state.

        Returns:
            Optional[int]: Time in seconds it took for the job to run. If the job is not in a terminal
                state then `None` is returned.
        """
        if self.job_start_time is not None and self.job_end_time is not None:
            return self.job_end_time - self.job_start_time
        return None

    def get_start_time(self) -> Optional[datetime]:
        """Get the time the job started running.

        Returns:
            Optional[datetime]: Datetime when the job started running. If the job has not started
                then `None` is returned.
        """
        if self.job_start_time is None:
            return None
        return datetime.utcfromtimestamp(self.job_start_time)

    def get_end_time(self) -> Optional[datetime]:
        """Get the time the job stopped running if the job is in a terminal state.

        Returns:
            Optional[datetime]: Datetime when the job started running. If the job is not in a terminal
                state then `None` is returned.
        """
        if self.job_end_time is None:
            return None
        return datetime.utcfromtimestamp(self.job_end_time)

    @classmethod
    def from_response(cls, response: Any) -> "BatchJobResult":
        """Construct a `BatchJobResult` from a batch API job response.

        Args:
            response (Any): Batch API job response.

        Returns:
            BatchJobResult: A `BatchJobResult` based on a batch API job response.
        """
        try:
            request = response["request"]
            configs = {}
            for model_name, config_dict in request["models"].items():
                model_type = ModelType.from_str(model_name)
                config = config_from_model_type(model_type).from_dict(config_dict)
                configs[model_type] = config

            kwargs = {}
            if "completed" in response:
                completed_dict = response["completed"]
                kwargs["artifacts_url"] = completed_dict["artifacts_url"]
                kwargs["errors_url"] = completed_dict["errors_url"]
                kwargs["predictions_url"] = completed_dict["predictions_url"]

            if "failed" in response:
                failed_dict = response["failed"]
                if "message" in failed_dict:
                    kwargs["error_message"] = failed_dict["message"]

            if "creation_timestamp" in response:
                kwargs["job_start_time"] = response["creation_timestamp"]

            if "completion_timestamp" in response:
                kwargs["job_end_time"] = response["completion_timestamp"]

            return cls(
                configs=configs,
                urls=request["urls"],
                status=BatchJobStatus.from_str(response["status"]),
                **kwargs,
            )
        # pylint: disable=broad-except
        except Exception as exc:
            message = cls._get_invalid_response_message(response)
            raise HumeClientException(message) from exc

    @classmethod
    def _get_invalid_response_message(cls, response: Any) -> str:
        response_str = json.dumps(response)
        message = f"Could not parse response into BatchJobResult: {response_str}"

        # Check for invalid API key
        if "fault" in response and "faultstring" in response["fault"]:
            fault_string = response["fault"]["faultstring"]
            if fault_string == "Invalid ApiKey":
                message = "HumeBatchClient initialized with invalid API key."

        return message

__init__(*, configs, urls, status, predictions_url=None, artifacts_url=None, errors_url=None, error_message=None, job_start_time=None, job_end_time=None)

Construct a BatchJobResult.

Parameters:

Name Type Description Default
configs Dict[ModelType, ModelConfigBase]

Configurations for the BatchJob.

 required
urls List[str]

URLs processed in the BatchJob.

 required
status BatchJobStatus

Status of BatchJob.

 required
predictions_url Optional[str]

URL to predictions file.

 None
artifacts_url Optional[str]

URL to artifacts zip archive.

 None
errors_url Optional[str]

URL to errors file.

 None
error_message Optional[str]

Error message for request.

 None
job_start_time Optional[int]

Time when job started.

 None
job_end_time Optional[int]

Time when job completed.

 None
Source code in hume/_batch/batch_job_result.py 
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
def __init__(
    self,
    *,
    configs: Dict[ModelType, ModelConfigBase],
    urls: List[str],
    status: BatchJobStatus,
    predictions_url: Optional[str] = None,
    artifacts_url: Optional[str] = None,
    errors_url: Optional[str] = None,
    error_message: Optional[str] = None,
    job_start_time: Optional[int] = None,
    job_end_time: Optional[int] = None,
):
    """Construct a BatchJobResult.

    Args:
        configs (Dict[ModelType, ModelConfigBase]): Configurations for the `BatchJob`.
        urls (List[str]): URLs processed in the `BatchJob`.
        status (BatchJobStatus): Status of `BatchJob`.
        predictions_url (Optional[str]): URL to predictions file.
        artifacts_url (Optional[str]): URL to artifacts zip archive.
        errors_url (Optional[str]): URL to errors file.
        error_message (Optional[str]): Error message for request.
        job_start_time (Optional[int]): Time when job started.
        job_end_time (Optional[int]): Time when job completed.
    """
    self.configs = configs
    self.urls = urls
    self.status = status
    self.predictions_url = predictions_url
    self.artifacts_url = artifacts_url
    self.errors_url = errors_url
    self.error_message = error_message
    self.job_start_time = job_start_time
    self.job_end_time = job_end_time

download_artifacts(filepath=None)

Download BatchJob artifacts zip archive.

Parameters:

Name Type Description Default
filepath Optional[Union[str, Path]]

Filepath where artifacts zip archive will be downloaded.

 None
Source code in hume/_batch/batch_job_result.py 
64
65
66
67
68
69
70
71
72
def download_artifacts(self, filepath: Optional[Union[str, Path]] = None) -> None:
    """Download `BatchJob` artifacts zip archive.

    Args:
        filepath (Optional[Union[str, Path]]): Filepath where artifacts zip archive will be downloaded.
    """
    if self.artifacts_url is None:
        raise HumeClientException("Could not download job artifacts. No artifacts found on job result.")
    urlretrieve(self.artifacts_url, filepath)

download_errors(filepath=None)

Download BatchJob errors file.

Parameters:

Name Type Description Default
filepath Optional[Union[str, Path]]

Filepath where errors will be downloaded.

 None
Source code in hume/_batch/batch_job_result.py 
74
75
76
77
78
79
80
81
82
def download_errors(self, filepath: Optional[Union[str, Path]] = None) -> None:
    """Download `BatchJob` errors file.

    Args:
        filepath (Optional[Union[str, Path]]): Filepath where errors will be downloaded.
    """
    if self.errors_url is None:
        raise HumeClientException("Could not download job errors. No errors found on job result.")
    urlretrieve(self.errors_url, filepath)

download_predictions(filepath=None)

Download BatchJob predictions file.

Parameters:

Name Type Description Default
filepath Optional[Union[str, Path]]

Filepath where predictions will be downloaded.

 None
Source code in hume/_batch/batch_job_result.py 
54
55
56
57
58
59
60
61
62
def download_predictions(self, filepath: Optional[Union[str, Path]] = None) -> None:
    """Download `BatchJob` predictions file.

    Args:
        filepath (Optional[Union[str, Path]]): Filepath where predictions will be downloaded.
    """
    if self.predictions_url is None:
        raise HumeClientException("Could not download job predictions. No predictions found on job result.")
    urlretrieve(self.predictions_url, filepath)

from_response(response) classmethod

Construct a BatchJobResult from a batch API job response.

Parameters:

Name Type Description Default
response Any

Batch API job response.

 required

Returns:

Name Type Description
BatchJobResult BatchJobResult

A BatchJobResult based on a batch API job response.
Source code in hume/_batch/batch_job_result.py 
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
@classmethod
def from_response(cls, response: Any) -> "BatchJobResult":
    """Construct a `BatchJobResult` from a batch API job response.

    Args:
        response (Any): Batch API job response.

    Returns:
        BatchJobResult: A `BatchJobResult` based on a batch API job response.
    """
    try:
        request = response["request"]
        configs = {}
        for model_name, config_dict in request["models"].items():
            model_type = ModelType.from_str(model_name)
            config = config_from_model_type(model_type).from_dict(config_dict)
            configs[model_type] = config

        kwargs = {}
        if "completed" in response:
            completed_dict = response["completed"]
            kwargs["artifacts_url"] = completed_dict["artifacts_url"]
            kwargs["errors_url"] = completed_dict["errors_url"]
            kwargs["predictions_url"] = completed_dict["predictions_url"]

        if "failed" in response:
            failed_dict = response["failed"]
            if "message" in failed_dict:
                kwargs["error_message"] = failed_dict["message"]

        if "creation_timestamp" in response:
            kwargs["job_start_time"] = response["creation_timestamp"]

        if "completion_timestamp" in response:
            kwargs["job_end_time"] = response["completion_timestamp"]

        return cls(
            configs=configs,
            urls=request["urls"],
            status=BatchJobStatus.from_str(response["status"]),
            **kwargs,
        )
    # pylint: disable=broad-except
    except Exception as exc:
        message = cls._get_invalid_response_message(response)
        raise HumeClientException(message) from exc

get_end_time()

Get the time the job stopped running if the job is in a terminal state.

Returns:

Type Description
Optional[datetime]

Optional[datetime]: Datetime when the job started running. If the job is not in a terminal state then None is returned.
Source code in hume/_batch/batch_job_result.py 
114
115
116
117
118
119
120
121
122
123
def get_end_time(self) -> Optional[datetime]:
    """Get the time the job stopped running if the job is in a terminal state.

    Returns:
        Optional[datetime]: Datetime when the job started running. If the job is not in a terminal
            state then `None` is returned.
    """
    if self.job_end_time is None:
        return None
    return datetime.utcfromtimestamp(self.job_end_time)

get_error_message()

Get any available error messages on the job.

Returns:

Type Description
Optional[str]

Optional[str]: A string with the error message if there was an error, otherwise None.
Source code in hume/_batch/batch_job_result.py 
84
85
86
87
88
89
90
def get_error_message(self) -> Optional[str]:
    """Get any available error messages on the job.

    Returns:
        Optional[str]: A string with the error message if there was an error, otherwise `None`.
    """
    return self.error_message

get_run_time()

Get the total time in seconds it took for the job to run if the job is in a terminal state.

Returns:

Type Description
Optional[int]

Optional[int]: Time in seconds it took for the job to run. If the job is not in a terminal state then None is returned.
Source code in hume/_batch/batch_job_result.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_run_time(self) -> Optional[int]:
    """Get the total time in seconds it took for the job to run if the job is in a terminal state.

    Returns:
        Optional[int]: Time in seconds it took for the job to run. If the job is not in a terminal
            state then `None` is returned.
    """
    if self.job_start_time is not None and self.job_end_time is not None:
        return self.job_end_time - self.job_start_time
    return None

get_start_time()

Get the time the job started running.

Returns:

Type Description
Optional[datetime]

Optional[datetime]: Datetime when the job started running. If the job has not started then None is returned.
Source code in hume/_batch/batch_job_result.py 
103
104
105
106
107
108
109
110
111
112
def get_start_time(self) -> Optional[datetime]:
    """Get the time the job started running.

    Returns:
        Optional[datetime]: Datetime when the job started running. If the job has not started
            then `None` is returned.
    """
    if self.job_start_time is None:
        return None
    return datetime.utcfromtimestamp(self.job_start_time)