rest api sample

Author: katzy687

generic-orchestration-samples/setup/setup_send_command/__main__.py

@@ -0,0 +1,9 @@
+from cloudshell.workflow.orchestration.setup.default_setup_orchestrator import DefaultSetupWorkflow
+from cloudshell.workflow.orchestration.sandbox import Sandbox
+from send_command import send_command
+
+sandbox = Sandbox()
+
+DefaultSetupWorkflow().register(sandbox)
+sandbox.workflow.on_configuration_ended(send_command, None)
+sandbox.execute_setup()

generic-orchestration-samples/setup/setup_send_command/requirements.txt

@@ -0,0 +1 @@
+cloudshell-orch-core>=3.4.0.0,<3.5.0.0

generic-orchestration-samples/setup/setup_send_command/send_command.py

@@ -0,0 +1,38 @@
+from cloudshell.workflow.orchestration.sandbox import Sandbox
+from cloudshell.api.cloudshell_api import InputNameValue
+
+
+LINUX_MODEL = "Linux Server"
+COMMAND_STRING = "/root/hello_world.sh"
+
+
+# ========== Primary Function ==========
+def send_command(sandbox, components=None):
+    """
+    Functions passed into orchestration flow MUST have (sandbox, components) signature
+    :param Sandbox sandbox:
+    :param components
+    :return:
+    """
+    api = sandbox.automation_api
+    res_id = sandbox.id
+    res_details = api.GetReservationDetails(res_id).ReservationDescription
+    resources = res_details.Resources
+    linux_resources = [resource for resource in resources
+                       if resource.ResourceModelName == LINUX_MODEL]
+    if not linux_resources:
+        raise Exception("No Linux Server found")
+
+    linux_resource = linux_resources[0]
+
+    command_inputs = [InputNameValue("command", COMMAND_STRING)]
+    output = api.ExecuteCommand(reservationId=res_id,
+                                targetName=linux_resource.Name,
+                                targetType="Resource",
+                                commandName="send_custom_command",
+                                commandInputs=command_inputs,
+                                printOutput=True).Output
+
+    # validate output
+    if "hello world" not in output.lower():
+        raise Exception(f"Command Validation failed. 'hello world' not found.\nOutput received: {output}")

generic-orchestration-samples/setup_call_health_check/__init__.py

@@ -0,0 +1,56 @@
+import json
+import sys
+import time
+
+from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession, InputParam
+
+if len(sys.argv) < 2:
+    raise ValueError("Sandbox Id not passed")
+
+# sandbox ID is injected into script via command line argument
+sandbox_id = sys.argv[1]
+
+# pull in api user credentials
+CS_SERVER = "localhost"
+CS_USER = "admin"
+CS_PASSWORD = "admin"
+CS_DOMAIN = "Global"
+
+TARGET_BLUEPRINT = "<MY_BLUEPRINT_NAME>"
+DEPLOYED_SANDBOX_NAME = "My Rest Api Blueprint"
+
+api = SandboxRestApiSession(host=CS_SERVER, username=CS_USER, password=CS_PASSWORD, domain=CS_DOMAIN)
+components_response = api.get_sandbox_components(sandbox_id)
+print("===== GETTING COMPONENT DETAILS IN SANDBOX =====")
+print(f"total components in sandbox: {len(components_response)}")
+print(f"components response:\n{json.dumps(components_response, indent=4)}")
+
+linux_server_search = [x for x in components_response if x["component_type"] == "Linux Server"]
+if not linux_server_search:
+    raise ValueError("linux server resource not found")
+
+linux_server = linux_server_search[0]
+linux_server_id = linux_server["id"]
+
+print("===== RUNNING COMMAND TO LINUX SERVER =====")
+params = [InputParam("command", "ifconfig")]
+command_response = api.run_component_command(sandbox_id=sandbox_id,
+                                             component_id=linux_server_id,
+                                             command_name="send_custom_command",
+                                             params=params)
+
+execution_id = command_response["executionId"]
+command_completed = False
+output = None
+while not command_completed:
+    details = api.get_execution_details(execution_id)
+    status = details["status"]
+    if status.lower() in ["completed", "failed"]:
+        command_completed = True
+        output = details["output"]
+    time.sleep(10)
+
+if output:
+    print(f"command output: {output}")
+
+print("sandbox api commands finished")