Skip to content

BatchJobDetails

Batch job details.

Source code in hume/_batch/batch_job_details.py 
 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
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
class BatchJobDetails:
    """Batch job details."""

    def __init__(
        self,
        *,
        configs: Dict[ModelType, ModelConfigBase],
        urls: List[str],
        files: List[str],
        state: BatchJobState,
        callback_url: Optional[str] = None,
        notify: bool = False,
    ):
        """Construct a BatchJobDetails.

        Args:
            configs (Dict[ModelType, ModelConfigBase]): Configurations for the `BatchJob`.
            urls (List[str]): URLs processed in the `BatchJob`.
            files (List[str]): Files processed in the `BatchJob`.
            state (BatchJobState): State of `BatchJob`.
            callback_url (Optional[str]): A URL to which a POST request is sent upon job completion.
            notify (bool): Whether an email notification should be sent upon job completion.
        """
        self.configs = configs
        self.urls = urls
        self.files = files
        self.state = state
        self.callback_url = callback_url
        self.notify = notify

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

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

        Returns:
            BatchJobDetails: A `BatchJobDetails` based on a batch API job response.
        """
        try:
            request = response["request"]

            configs = {}
            for model_name, config_dict in request["models"].items():
                if config_dict is None:
                    continue
                model_type = ModelType.from_str(model_name)
                config = config_from_model_type(model_type).from_dict(config_dict)
                configs[model_type] = config

            urls = request["urls"]
            files = request["files"]
            callback_url = request["callback_url"]
            notify = request["notify"]

            state_dict = response["state"]
            state = BatchJobState(
                status=BatchJobStatus.from_str(state_dict["status"]),
                created_timestamp_ms=state_dict.get("created_timestamp_ms"),
                started_timestamp_ms=state_dict.get("started_timestamp_ms"),
                ended_timestamp_ms=state_dict.get("ended_timestamp_ms"),
            )

            return cls(
                configs=configs,
                urls=urls,
                files=files,
                state=state,
                callback_url=callback_url,
                notify=notify,
            )
        # 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 BatchJobDetails: {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

    def get_status(self) -> BatchJobStatus:
        """Get the status of the job.

        Returns:
            BatchJobStatus: The status of the `BatchJob`.
        """
        return self.state.status

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

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

    def get_created_time(self) -> Optional[datetime]:
        """Get the time the job was created.

        Returns:
            Optional[datetime]: Datetime when the job was created. If the job has not started
                then `None` is returned.
        """
        if self.state.created_timestamp_ms is None:
            return None
        return datetime.utcfromtimestamp(self.state.created_timestamp_ms / 1000)

    def get_started_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.state.started_timestamp_ms is None:
            return None
        return datetime.utcfromtimestamp(self.state.started_timestamp_ms / 1000)

    def get_ended_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.state.ended_timestamp_ms is None:
            return None
        return datetime.utcfromtimestamp(self.state.ended_timestamp_ms / 1000)

__init__(*, configs, urls, files, state, callback_url=None, notify=False)

Construct a BatchJobDetails.

Parameters:

Name Type Description Default
configs Dict[ModelType, ModelConfigBase]

Configurations for the BatchJob.

 required
urls List[str]

URLs processed in the BatchJob.

 required
files List[str]

Files processed in the BatchJob.

 required
state BatchJobState

State of BatchJob.

 required
callback_url Optional[str]

A URL to which a POST request is sent upon job completion.

 None
notify bool

Whether an email notification should be sent upon job completion.

 False
Source code in hume/_batch/batch_job_details.py 
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
def __init__(
    self,
    *,
    configs: Dict[ModelType, ModelConfigBase],
    urls: List[str],
    files: List[str],
    state: BatchJobState,
    callback_url: Optional[str] = None,
    notify: bool = False,
):
    """Construct a BatchJobDetails.

    Args:
        configs (Dict[ModelType, ModelConfigBase]): Configurations for the `BatchJob`.
        urls (List[str]): URLs processed in the `BatchJob`.
        files (List[str]): Files processed in the `BatchJob`.
        state (BatchJobState): State of `BatchJob`.
        callback_url (Optional[str]): A URL to which a POST request is sent upon job completion.
        notify (bool): Whether an email notification should be sent upon job completion.
    """
    self.configs = configs
    self.urls = urls
    self.files = files
    self.state = state
    self.callback_url = callback_url
    self.notify = notify

from_response(response) classmethod

Construct a BatchJobDetails from a batch API job response.

Parameters:

Name Type Description Default
response Any

Batch API job response.

 required

Returns:

Name Type Description
BatchJobDetails BatchJobDetails

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

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

    Returns:
        BatchJobDetails: A `BatchJobDetails` based on a batch API job response.
    """
    try:
        request = response["request"]

        configs = {}
        for model_name, config_dict in request["models"].items():
            if config_dict is None:
                continue
            model_type = ModelType.from_str(model_name)
            config = config_from_model_type(model_type).from_dict(config_dict)
            configs[model_type] = config

        urls = request["urls"]
        files = request["files"]
        callback_url = request["callback_url"]
        notify = request["notify"]

        state_dict = response["state"]
        state = BatchJobState(
            status=BatchJobStatus.from_str(state_dict["status"]),
            created_timestamp_ms=state_dict.get("created_timestamp_ms"),
            started_timestamp_ms=state_dict.get("started_timestamp_ms"),
            ended_timestamp_ms=state_dict.get("ended_timestamp_ms"),
        )

        return cls(
            configs=configs,
            urls=urls,
            files=files,
            state=state,
            callback_url=callback_url,
            notify=notify,
        )
    # pylint: disable=broad-except
    except Exception as exc:
        message = cls._get_invalid_response_message(response)
        raise HumeClientException(message) from exc

get_created_time()

Get the time the job was created.

Returns:

Type Description
Optional[datetime]

Optional[datetime]: Datetime when the job was created. If the job has not started then None is returned.
Source code in hume/_batch/batch_job_details.py 
123
124
125
126
127
128
129
130
131
132
def get_created_time(self) -> Optional[datetime]:
    """Get the time the job was created.

    Returns:
        Optional[datetime]: Datetime when the job was created. If the job has not started
            then `None` is returned.
    """
    if self.state.created_timestamp_ms is None:
        return None
    return datetime.utcfromtimestamp(self.state.created_timestamp_ms / 1000)

get_ended_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_details.py 
145
146
147
148
149
150
151
152
153
154
def get_ended_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.state.ended_timestamp_ms is None:
        return None
    return datetime.utcfromtimestamp(self.state.ended_timestamp_ms / 1000)

get_run_time_ms()

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

Returns:

Type Description
Optional[int]

Optional[int]: Time in milliseconds 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_details.py 
112
113
114
115
116
117
118
119
120
121
def get_run_time_ms(self) -> Optional[int]:
    """Get the total time in milliseconds it took for the job to run if the job is in a terminal state.

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

get_started_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_details.py 
134
135
136
137
138
139
140
141
142
143
def get_started_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.state.started_timestamp_ms is None:
        return None
    return datetime.utcfromtimestamp(self.state.started_timestamp_ms / 1000)

get_status()

Get the status of the job.

Returns:

Name Type Description
BatchJobStatus BatchJobStatus

The status of the BatchJob.
Source code in hume/_batch/batch_job_details.py 
104
105
106
107
108
109
110
def get_status(self) -> BatchJobStatus:
    """Get the status of the job.

    Returns:
        BatchJobStatus: The status of the `BatchJob`.
    """
    return self.state.status