Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@check_no_op
@check_offline
def _end(self):
self.log_succeeded()
self._wait()
@check_no_op
@check_offline
def log_data_ref(self, data, data_name="data", reset=False):
try:
params = {data_name: hash_value(data)}
patch_dict = {"input": params}
if reset is False:
patch_dict["merge"] = True
self._update(patch_dict)
except Exception as e:
logger.warning("Could create data hash %s", e)
@check_no_op
def log_artifacts(self, dir_path):
self.outputs_store.upload_dir(dirname=dir_path)
@check_no_op
@check_offline
def set_name(self, name):
self._update({"name": name})
@check_no_op
@check_offline
def end(self, status, message=None, traceback=None):
if self.last_status in ["succeeded", "failed", "stopped"]:
return
self.log_status(status=status, message=message, traceback=traceback)
self.last_status = status
time.sleep(
0.1
) # Just to give the opportunity to the worker to pick the message
self._unset_health_url()
@check_no_op
@check_offline
def log_inputs(self, reset=False, **inputs):
patch_dict = {"inputs": inputs}
if reset is False:
patch_dict["merge"] = True
self.data.inputs = inputs
else:
self.data.inputs.update(inputs)
self._update(patch_dict)
@check_no_op
@check_offline
def _start(self):
atexit.register(self._end)
self.start()
def excepthook(exception, value, tb):
self.log_failed(message="Type: {}, Value: {}".format(exception, value))
# Resume normal work
sys.__excepthook__(exception, value, tb)
sys.excepthook = excepthook
@check_no_op
@check_offline
def refresh_data(self):
self._run = self.client.runs_v1.get_run(self.owner, self.project, self.run_uuid)
for status, conditions in get_run_statuses(
self.owner, self.project, self.run_uuid
):
self.status = status
@check_no_op
def set_outputs_store(
self, outputs_store=None, outputs_path=None, set_env_vars=False
):
if not any([outputs_store, outputs_path]):
raise PolyaxonClientException(
"An Store instance or and outputs path is required."
)
self.outputs_store = outputs_store or StoreManager(path=outputs_path)
if self.outputs_store and set_env_vars:
self.outputs_store.set_env_vars()