ProgressStatus
airt.client.ProgressStatus
A base class for querying status of a remote operation.
__init__(self, relative_url, sleep_for=5, timeout=0)
special
Constructs a new ProgressStatus instance.
Warning
Do not construct this object directly by calling the constructor, please use either progress_bar,
is_ready, or wait methods of DataBlob
, DataSource
, Model
or Prediction
classes instead.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
relative_url |
str |
Rest API endpoint relative url. |
required |
sleep_for |
Union[int, float] |
The time interval in seconds between successive API calls. |
5 |
timeout |
int |
The maximum time allowed in seconds for the asynchronous call to complete. If not the progressbar will be terminated. |
0 |
Exceptions:
Type | Description |
---|---|
TimeoutError |
in case of connection timeout. |
Source code in airt/client.py
def __init__(
self, relative_url: str, sleep_for: Union[int, float] = 5, timeout: int = 0
):
"""Constructs a new ProgressStatus instance.
Warning:
Do not construct this object directly by calling the constructor, please use either progress_bar,
is_ready, or wait methods of `DataBlob`, `DataSource`, `Model` or `Prediction` classes instead.
Args:
relative_url: Rest API endpoint relative url.
sleep_for: The time interval in seconds between successive API calls.
timeout: The maximum time allowed in seconds for the asynchronous call to complete. If not the
progressbar will be terminated.
Raises:
TimeoutError: in case of connection timeout.
"""
self.relative_url = relative_url
self.sleep_for = sleep_for
self.timeout = timeout
is_ready(self)
Check if the method's progress is complete.
Returns:
Type | Description |
---|---|
bool |
True if the progress if completed, else False. |
Source code in airt/client.py
def is_ready(self) -> bool:
"""Check if the method's progress is complete.
Returns:
**True** if the progress if completed, else **False**.
"""
response = Client._get_data(relative_url=self.relative_url)
return response["completed_steps"] == response["total_steps"]
progress_bar(self)
Blocks the execution and displays a progress bar showing the remote action progress.
Exceptions:
Type | Description |
---|---|
ConnectionError |
If the server address is invalid or not reachable. |
TimeoutError |
in case of connection timeout. |
Source code in airt/client.py
def progress_bar(self):
"""Blocks the execution and displays a progress bar showing the remote action progress.
Raises:
ConnectionError: If the server address is invalid or not reachable.
TimeoutError: in case of connection timeout.
"""
total_steps = Client._get_data(relative_url=self.relative_url)["total_steps"]
with tqdm(total=total_steps) as pbar:
started_at = datetime.now()
while True:
if (0 < self.timeout) and (datetime.now() - started_at) > timedelta(
seconds=self.timeout
):
raise TimeoutError()
response = Client._get_data(relative_url=self.relative_url)
completed_steps = response["completed_steps"]
pbar.update(completed_steps)
if completed_steps == total_steps:
break
sleep(self.sleep_for)
wait(self)
Blocks execution while waiting for the remote action to complete.
Exceptions:
Type | Description |
---|---|
ConnectionError |
If the server address is invalid or not reachable. |
TimeoutError |
in case of timeout. |
Source code in airt/client.py
@patch
def wait(self: ProgressStatus):
"""Blocks execution while waiting for the remote action to complete.
Raises:
ConnectionError: If the server address is invalid or not reachable.
TimeoutError: in case of timeout.
"""
started_at = datetime.now()
while True:
if (0 < self.timeout) and (datetime.now() - started_at) > timedelta(
seconds=self.timeout
):
raise TimeoutError()
if self.is_ready():
return
sleep(self.sleep_for)