Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion python/v4api_sdk/batch_ops_actions.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@
from ntnx_prism_py_client.models.prism.v4.operations.ActionType import ActionType


from tme import Utils
from tme.utils import Utils


def confirm_entity(api, client, entity_name: str) -> str:
Expand Down
2 changes: 1 addition & 1 deletion python/v4api_sdk/batch_ops_create.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@

from ntnx_prism_py_client.models.prism.v4.operations.ActionType import ActionType

from tme import Utils
from tme.utils import Utils


def main():
Expand Down
2 changes: 1 addition & 1 deletion python/v4api_sdk/batch_ops_modify.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@

from ntnx_prism_py_client.models.prism.v4.operations.ActionType import ActionType

from tme import Utils
from tme.utils import Utils


def main():
Expand Down
2 changes: 1 addition & 1 deletion python/v4api_sdk/create_categories.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
from ntnx_prism_py_client.rest import ApiException as PrismException


from tme import Utils
from tme.utils import Utils
from tme import Config


Expand Down
5 changes: 2 additions & 3 deletions python/v4api_sdk/create_image_sdk.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
from ntnx_clustermgmt_py_client import ApiClient as ClusterClient
from ntnx_clustermgmt_py_client import Configuration as ClusterConfiguration

from tme import Utils
from tme.utils import Utils


def main():
Expand Down Expand Up @@ -46,8 +46,7 @@ def main():

# get the cluster password
cluster_password = getpass.getpass(
prompt="Please enter your Prism Central \
password: ",
prompt="Please enter your Prism Central password: ",
stream=None,
)

Expand Down
2 changes: 1 addition & 1 deletion python/v4api_sdk/create_network_security_policy.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
from ntnx_prism_py_client.rest import ApiException as PrismException


from tme import Utils
from tme.utils import Utils
from tme import Config


Expand Down
6 changes: 3 additions & 3 deletions python/v4api_sdk/create_subnet_sdk.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,8 +36,8 @@
from ntnx_prism_py_client.rest import ApiException as PrismException


from tme import Utils
from tme import Config
from tme.utils import Utils
from tme.utils import Config


def confirm_entity(api, client, entity_name: str, exclusions: list) -> str:
Expand Down Expand Up @@ -297,7 +297,7 @@ def main():

# did the user say Yes to creating the subnet?
if confirm_create:
networking_instance = ntnx_networking_py_client.api.SubnetApi(
networking_instance = ntnx_networking_py_client.api.SubnetsApi(
api_client=networking_client
)

Expand Down
2 changes: 1 addition & 1 deletion python/v4api_sdk/create_vm_sdk.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@
from ntnx_networking_py_client import Configuration as NetworkingConfiguration
from ntnx_networking_py_client import ApiClient as NetworkingClient

from tme import Utils
from tme.utils import Utils


def confirm_entity(api, client, entity_name: str, exclusions: list) -> str:
Expand Down
2 changes: 2 additions & 0 deletions python/v4api_sdk/dr/.gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
.DS_Store
venv/
16 changes: 8 additions & 8 deletions python/v4api_sdk/requirements.txt
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
requests==2.31.0
ruff==0.1.15
ntnx_vmm_py_client==4.0.1b1
ntnx_lifecycle_py_client==4.0.1b1
ntnx_prism_py_client==4.0.1b1
ntnx_clustermgmt_py_client==4.0.1b2
ntnx_aiops_py_client==4.0.3a2
ntnx_networking_py_client==4.0.1b1
ntnx_microseg_py_client==4.0.1b1
ntnx_iam_py_client==4.0.1b1
ntnx_vmm_py_client==4.0.1
ntnx_lifecycle_py_client==4.0.1
ntnx_prism_py_client==4.0.1
ntnx_clustermgmt_py_client==4.0.1
ntnx_aiops_py_client==4.0.1
ntnx_networking_py_client==4.0.1
ntnx_microseg_py_client==4.0.1
ntnx_iam_py_client==4.0.1
114 changes: 0 additions & 114 deletions python/v4api_sdk/tme/__init__.py
Original file line number Diff line number Diff line change
@@ -1,114 +0,0 @@
"""
Simple module to allow function re-use across Nutanix
v4 SDK code samples

Requires Prism Central 2024.1 or later, AOS 6.8 or later
"""

import time
import urllib3
from dataclasses import dataclass
from timeit import default_timer as timer

import ntnx_prism_py_client
from ntnx_prism_py_client import ApiClient as PrismClient
from ntnx_prism_py_client import Configuration as PrismConfiguration


@dataclass
class Config:
"""
dataclass to hold configuration for each script run
nice and modular
"""

pc_ip: str
pc_username: str
pc_password: str

class Utils:
"""
class to manage simple reusable functions across the Python
v4 API code samples
"""

prism_config: PrismConfiguration

def __init__(self, pc_ip: str, username: str, password: str):
"""
class constructor
create reusable instances of Prism connections (etc)
"""
self.prism_config = PrismConfiguration()
self.prism_config.host = pc_ip
self.prism_config.username = username
self.prism_config.password = password
self.prism_config.verify_ssl = False
self.prism_client = PrismClient(configuration=self.prism_config)
self.prism_client.add_default_header(
header_name="Accept-Encoding", header_value="gzip, deflate, br"
)
self.prism_instance = ntnx_prism_py_client.api.TasksApi(
api_client=self.prism_client
)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_task(self, ext_id: str):
"""
method used to get details of a specified task
"""
task = self.prism_instance.get_task_by_id(f"{ext_id}")
if task:
return task
return None

def confirm(self, message: str):
"""
method to request a yes/NO confirmation from the user
used to run or skip precheck, inventory (etc)
"""
yes_no = input(f"{message} (yes/NO): ").lower()
return yes_no == "yes"

def monitor_task(
self, task_ext_id, task_name, pc_ip, username, password, poll_timeout, prefix = ""
):
"""
method used to monitor Prism Central tasks
will print a series of period characters and re-check task
status at the specified interval
this version uses the Prism SDK
"""
start = timer()
# print message until specified task is finished
prism_config = PrismConfiguration()
prism_config.host = pc_ip
prism_config.username = username
prism_config.password = password
prism_config.verify_ssl = False
prism_client = PrismClient(configuration=prism_config)
prism_client.add_default_header(
header_name="Accept-Encoding", header_value="gzip, deflate, br"
)
prism_instance = ntnx_prism_py_client.api.TasksApi(api_client=prism_client)
task = prism_instance.get_task_by_id(f"{prefix}{task_ext_id}")
units = "second" if poll_timeout == 1 else "seconds"
print(
f"{task_name} running, checking progress every {poll_timeout} {units} ...",
end="",
)
while True:
if task.data.status == "RUNNING":
print(".", end="", flush=True)
else:
print(" finished.")
break
time.sleep(int(poll_timeout))
task = prism_instance.get_task_by_id(f"{prefix}{task_ext_id}")
end = timer()
elapsed_time = end - start
if elapsed_time <= 60:
duration = f"{round(elapsed_time, 0)} seconds"
else:
duration = f"{round(elapsed_time // 60, 0)} minutes"
return duration
15 changes: 15 additions & 0 deletions python/v4api_sdk/tme/test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
import unittest
from utils import Utils

class TestUtils(unittest.TestCase):

def test_confirm_yes(self):
utils = Utils()
self.assertEqual(utils.confirm("Enter yes:"), True)

def test_confirm_no(self):
utils = Utils()
self.assertEqual(utils.confirm("Enter no:"), False)

if __name__ == "__main__":
unittest.main()
118 changes: 118 additions & 0 deletions python/v4api_sdk/tme/utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
"""
Simple module to allow function re-use across Nutanix
v4 SDK code samples

Requires Prism Central 2024.1 or later, AOS 6.8 or later
"""

import time
import urllib3
from dataclasses import dataclass
from timeit import default_timer as timer

import ntnx_prism_py_client
from ntnx_prism_py_client import ApiClient as PrismClient
from ntnx_prism_py_client import Configuration as PrismConfiguration


@dataclass
class Config:
"""
dataclass to hold configuration for each script run
nice and modular
"""

pc_ip: str
pc_username: str
pc_password: str

class Utils:
"""
class to manage simple reusable functions across the Python
v4 API code samples
"""

prism_config: PrismConfiguration

def __init__(self, pc_ip: str, username: str, password: str):
"""
class constructor
create reusable instances of Prism connections (etc)
"""
self.prism_config = PrismConfiguration()
self.prism_config.host = pc_ip
self.prism_config.username = username
self.prism_config.password = password
self.prism_config.verify_ssl = False
self.prism_client = PrismClient(configuration=self.prism_config)
self.prism_client.add_default_header(
header_name="Accept-Encoding", header_value="gzip, deflate, br"
)
self.prism_instance = ntnx_prism_py_client.api.TasksApi(
api_client=self.prism_client
)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_task(self, ext_id: str):
"""
method used to get details of a specified task
"""
task = self.prism_instance.get_task_by_id(f"{ext_id}")
if task:
return task
return None

def confirm(self, message: str):
"""
method to request a yes/NO confirmation from the user
used to run or skip precheck, inventory (etc)
"""
yes_no = input(f"{message} (yes/NO): ").lower()
return yes_no == "yes"

def monitor_task(
self, task_ext_id, task_name, pc_ip, username, password, poll_timeout, prefix = ""
):
"""
method used to monitor Prism Central tasks
will print a series of period characters and re-check task
status at the specified interval
this version uses the Prism SDK
"""
start = timer()
# print message until specified task is finished
prism_config = PrismConfiguration()
prism_config.host = pc_ip
prism_config.username = username
prism_config.password = password
prism_config.verify_ssl = False
prism_client = PrismClient(configuration=prism_config)
prism_client.add_default_header(
header_name="Accept-Encoding", header_value="gzip, deflate, br"
)
prism_instance = ntnx_prism_py_client.api.TasksApi(api_client=prism_client)
task = prism_instance.get_task_by_id(f"{prefix}{task_ext_id}")
units = "second" if poll_timeout == 1 else "seconds"
print(
f"{task_name} running, checking progress every {poll_timeout} {units} (progress will update when percentage complete changes) ...",
end="",
)
percent_complete = 0
print(f" {task.data.progress_percentage}% ... ", end="", flush=True)
while True:
if task.data.status == "RUNNING":
if task.data.progress_percentage > percent_complete:
print(f" {task.data.progress_percentage}% ... ", end="", flush=True)
percent_complete = task.data.progress_percentage
else:
print("finished.")
break
time.sleep(int(poll_timeout))
task = prism_instance.get_task_by_id(f"{prefix}{task_ext_id}")
end = timer()
elapsed_time = end - start
if elapsed_time <= 60:
duration = f"{round(elapsed_time, 0)} seconds"
else:
duration = f"{round(elapsed_time // 60, 0)} minutes"
return duration
2 changes: 1 addition & 1 deletion python/v4api_sdk/update_image_etag_sdk.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
from ntnx_prism_py_client import ApiClient as PrismClient
from ntnx_prism_py_client import Configuration as PrismConfiguration

from tme import Utils
from tme.utils import Utils

"""
suppress warnings about insecure connections
Expand Down