Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_node_info(node_config):
if not node_config:
Printer.print_error("No node was found.")
sys.exit(1)
node = node_config.to_dict()
node.pop("gpus")
Printer.print_header("Node info:")
dict_tabulate(Printer.add_memory_unit(node, "memory"))
gpus_items = list_dicts_to_tabulate(
[
Printer.add_memory_unit(gpu.to_light_dict(), "memory")
for gpu in node_config.gpus
]
)
if gpus_items:
Printer.print_header("Node GPUs:")
dict_tabulate(gpus_items, is_list_dict=True)
"""Delete a user.
Example:
\b
```bash
$ polyaxon user delete david
```
"""
try:
PolyaxonClient().user.delete_user(username)
except (PolyaxonHTTPError, PolyaxonShouldExitError, PolyaxonClientException) as e:
handle_cli_error(e, message="Could not delete user `{}`.".format(username))
sys.exit(1)
Printer.print_success("User `{}` was deleted successfully.".format(username))
parsed_params = parse_params(params)
if not any(exists):
Printer.print_error(
"Polyaxonfile is not present, "
"please run {}".format(constants.INIT_COMMAND)
)
sys.exit(1)
try:
plx_file = PolyaxonFile(polyaxonfile)
plx_file = plx_file.get_op_specification(
params=parsed_params, profile=profile, queue=queue, nocache=nocache
)
if log:
Printer.print_success("Polyaxonfile valid")
return plx_file
except Exception as e:
handle_cli_error(e, message="Polyaxonfile is not valid.")
sys.exit(1)
def set(**kwargs): # pylint:disable=redefined-builtin
"""Set the global config values.
Example:
\b
```bash
$ polyaxon config set --host=localhost
```
"""
try:
_config = ClientConfigManager.get_config_or_default()
except Exception as e:
handle_cli_error(e, message="Polyaxon load configuration.")
Printer.print_header(
"You can reset your config by running: polyaxon config purge"
)
sys.exit(1)
for key, value in kwargs.items():
if value is not None:
setattr(_config, key, value)
ClientConfigManager.set_config(_config)
Printer.print_success("Config was updated.")
# Reset cli config
CliConfigManager.purge()
def create_debug_polyaxonfile():
if os.path.isfile(constants.DEBUG_FILE_PATH):
Printer.print_success("A polyaxonfile.debug.yaml was found in the project.")
else:
create_debug_file(constants.DEBUG_FILE)
# if we are here the file was not created
if not os.path.isfile(constants.DEBUG_FILE_PATH):
Printer.print_error(
"Something went wrong, init command did not create a debug file.\n"
"Possible reasons: you don't have enough rights to create the file."
)
sys.exit(1)
Printer.print_success(
"{} was created successfully.".format(constants.DEBUG_FILE_PATH)
)
def create_run():
click.echo("Creating a run.")
run = V1Run(content=specification.config_dump)
try:
polyaxon_client = PolyaxonClient()
response = polyaxon_client.runs_v1.create_run(owner, project_name, run)
cache.cache(config_manager=RunManager, response=response)
Printer.print_success("A new run `{}` was created".format(response.uuid))
except (ApiException, HTTPError) as e:
handle_cli_error(e, message="Could not create a run.")
sys.exit(1)
polyaxon_client = PolyaxonClient()
response = polyaxon_client.runs_v1.list_bookmarked_runs(user, **params)
except (ApiException, HTTPError) as e:
handle_cli_error(
e,
message="Could not get bookmarked experiments for user `{}`.".format(user),
)
sys.exit(1)
meta = get_meta_response(response)
if meta:
Printer.print_header("Bookmarked experiments for user `{}`.".format(user))
Printer.print_header("Navigation:")
dict_tabulate(meta)
else:
Printer.print_header(
"No bookmarked experiments found for user `{}`.".format(user)
)
objects = [
Printer.add_status_color(o.to_light_dict(humanize_values=True))
for o in response.results
]
objects = list_dicts_to_tabulate(objects)
if objects:
Printer.print_header("Experiments:")
dict_tabulate(objects, is_list_dict=True)
_run(
ctx,
name,
owner,
project_name,
description,
tags,
specification,
log,
conda_env,
)
except (PolyaxonHTTPError, PolyaxonShouldExitError, PolyaxonClientException) as e:
handle_cli_error(e, message="Could start local run.")
sys.exit(1)
except Exception as e:
Printer.print_error("Could start local run.")
Printer.print_error("Unexpected Error: `{}`.".format(e))
sys.exit(1)
def get_logs_handler(handle_job_info=False, show_timestamp=True, stream=True):
colors = deque(Printer.COLORS)
job_to_color = {}
def get_job_info(log_line, job_regex):
log_search = job_regex.search(log_line)
if not log_search:
return log_line, False
job_info = log_search.group()
if job_info in job_to_color:
color = job_to_color[job_info]
else:
color = colors[0]
colors.rotate(-1)
job_to_color[job_info] = color
return re.sub(job_regex, Printer.add_color(job_info, color), log_line), True
def get_run_details(run): # pylint:disable=redefined-outer-name
if run.description:
Printer.print_header("Run description:")
click.echo("{}\n".format(run.description))
if run.inputs:
Printer.print_header("Run inputs:")
dict_tabulate(run.inputs)
if run.outputs:
Printer.print_header("Run outputs:")
dict_tabulate(run.outputs)
response = Printer.add_status_color(run.to_dict())
response = dict_to_tabulate(
response,
humanize_values=True,
exclude_attrs=[
"project",