split _follow_deployment into two methods
@@ -671,9 +671,10 @@ class VespaCloud(object):
print(response["message"])
return response["run"]
- def _follow_deployment(self, instance: str, job: str, run: int):
- last = -1
- while True:
+ def _get_deployment_status(
+ self, instance: str, job: str, run: int, last: int
+ ) -> (str, int):
+
update = self._request(
"GET",
"/application/v4/tenant/{}/application/{}/instance/{}/job/{}/run/{}?after={}".format(
@@ -687,11 +688,11 @@ class VespaCloud(object):
last = update.get("lastId", last)
if update["active"]:
- sleep(1)
+ return "active", last
else:
status = update["status"]
if status == "success":
- return
+ return "success", last
elif status == "error":
raise RuntimeError(
"Unexpected error during deployment; see log for details"
@@ -721,6 +722,21 @@ class VespaCloud(object):
else:
raise RuntimeError("Unexpected status '" + status + "'")
+ def _follow_deployment(self, instance: str, job: str, run: int) -> None:
+ last = -1
+ while True:
+ try:
+ status, last = self._get_deployment_status(instance, job, run, last)
+ except RuntimeError:
+ raise
+
+ if status == "active":
+ sleep(1)
+ elif status == "success":
+ return
+ else:
+ raise NotImplementedError
+
@staticmethod
def _print_log_entry(step: str, entry: dict):
timestamp = strftime("%H:%M:%S", gmtime(entry["at"] / 1e3))
