212 lines
8.8 KiB
Python
212 lines
8.8 KiB
Python
"""interactive_cli.py
|
||
An application‑specific, context‑aware CLI for xCloudify.
|
||
|
||
Key capabilities
|
||
================
|
||
1. **Domain‑aware menus** – Presents high‑level objects (Workloads, Pods, Virtual_Data_Centers, etc.) instead of raw paths.
|
||
2. **Context memory** – Once you select an item (e.g. a Workload), subsequent actions are scoped to it until you navigate back.
|
||
3. **Action shortcuts** – For each object type the CLI surfaces common verbs (start/stop/restart for Workloads, attach/detach for Volumes, …) and calls the correct API for you.
|
||
4. **Rich TUI UX** – Uses *prompt_toolkit* for arrow‑key navigation, search‑as‑you‑type, history, and coloured output. Falls back to simple `input()` if the library is unavailable.
|
||
5. **Re‑usable Client** – Wraps the existing `IaaSClient` but adds automatic pagination & pretty printing.
|
||
|
||
Run with:
|
||
```
|
||
pip install requests prompt_toolkit tabulate
|
||
python interactive_cli.py --api-key <KEY> --base-url http://localhost:5000/api
|
||
```
|
||
|
||
You can start the CLI, hit `Workloads → List`, arrow‑down to pick one, hit *Enter* and get a submenu: `Details`, `Start`, `Stop`, `Restart`, `Delete`, `Show Volumes`, etc.
|
||
|
||
Full code below ⬇️
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import logging
|
||
import sys
|
||
from typing import Callable, Dict, List, Optional
|
||
|
||
import requests
|
||
from tabulate import tabulate
|
||
|
||
try:
|
||
from prompt_toolkit import PromptSession
|
||
from prompt_toolkit.completion import Completer, Completion
|
||
from prompt_toolkit.styles import Style
|
||
_PROMPT_AVAILABLE = True
|
||
except ImportError: # graceful fallback
|
||
_PROMPT_AVAILABLE = False
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Logging setup #
|
||
# --------------------------------------------------------------------------- #
|
||
LOG_FORMAT = "% (asctime)s | %(levelname)s | %(message)s"
|
||
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
|
||
logger = logging.getLogger("xcloudify_cli")
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# HTTP Client wrapper #
|
||
# --------------------------------------------------------------------------- #
|
||
class ApiClient:
|
||
"""Light wrapper around the IaaS REST API with basic error handling."""
|
||
|
||
def __init__(self, base_url: str, api_key: str) -> None:
|
||
self.base_url = base_url.rstrip("/")
|
||
self.session = requests.Session()
|
||
self.session.headers.update({"X-Internal-Xcloudify-API": api_key, "Content-Type": "application/json"})
|
||
|
||
# Generic helpers -------------------------------------------------------- #
|
||
def _url(self, path: str) -> str:
|
||
return f"{self.base_url}/{path.lstrip('/')}"
|
||
|
||
def _req(self, method: str, path: str, **kwargs):
|
||
url = self._url(path)
|
||
logger.debug("%s %s", method.upper(), url)
|
||
response = self.session.request(method, url, **kwargs)
|
||
if not response.ok:
|
||
logger.error("HTTP %s → %s %s", method.upper(), response.status_code, response.text)
|
||
response.raise_for_status()
|
||
return response.json() if response.headers.get("Content-Type", "").startswith("application/json") else response.text
|
||
|
||
# Workload endpoints ----------------------------------------------------- #
|
||
def list_workloads(self) -> List[Dict]:
|
||
"""Return all non‑deleted workloads (containers + Virtual_Machines)."""
|
||
return self._req("get", "/workloads")
|
||
|
||
def get_workload(self, workload_id: str) -> Dict:
|
||
return self._req("get", f"/workloads/containers/{workload_id}")
|
||
|
||
def workload_action(self, workload_id: str, action: str) -> Dict:
|
||
return self._req("post", f"/workloads/containers/{workload_id}/lifecycle/{action}")
|
||
|
||
# Extend for pods, volumes, networks as needed …
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# prompt_toolkit Completer #
|
||
# --------------------------------------------------------------------------- #
|
||
class ListCompleter(Completer):
|
||
"""Simple completer that offers a static list of words."""
|
||
|
||
def __init__(self, words: List[str]):
|
||
self.words = words
|
||
|
||
def get_completions(self, document, complete_event):
|
||
word = document.text_before_cursor.lower()
|
||
for w in self.words:
|
||
if w.lower().startswith(word):
|
||
yield Completion(w, start_position=-len(word))
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# CLI Controller #
|
||
# --------------------------------------------------------------------------- #
|
||
class XCloudifyCLI:
|
||
"""Interactive, context‑aware command‑line interface for xCloudify."""
|
||
|
||
MAIN_MENU = [
|
||
"Workloads",
|
||
"Pods",
|
||
"Virtual_Data_Centers",
|
||
"Regions",
|
||
"Images",
|
||
"Exit",
|
||
]
|
||
|
||
ACTIONS_WORKLOAD = [
|
||
"Details",
|
||
"Start",
|
||
"Stop",
|
||
"Restart",
|
||
"Delete",
|
||
"Back",
|
||
]
|
||
|
||
def __init__(self, client: ApiClient):
|
||
self.client = client
|
||
self.session: Optional[PromptSession] = None
|
||
if _PROMPT_AVAILABLE:
|
||
style = Style.from_dict({"prompt": "bold", "": ""})
|
||
self.session = PromptSession(style=style)
|
||
|
||
# ------------------------------ helpers -------------------------------- #
|
||
def _prompt(self, message: str, choices: List[str]) -> str:
|
||
"""Prompt the user to pick from *choices* (arrow‑keys if PTK, else numbered)."""
|
||
if self.session:
|
||
completer = ListCompleter(choices)
|
||
while True:
|
||
answer = self.session.prompt([("class:p", message + " ")], completer=completer).strip()
|
||
if answer in choices:
|
||
return answer
|
||
print("Invalid choice. Try again.")
|
||
else: # fallback
|
||
print(message)
|
||
for idx, choice in enumerate(choices, 1):
|
||
print(f" {idx}. {choice}")
|
||
while True:
|
||
sel = input("Select → ").strip()
|
||
if sel.isdigit() and 1 <= int(sel) <= len(choices):
|
||
return choices[int(sel) - 1]
|
||
print("Invalid selection. Try again.")
|
||
|
||
def _pretty(self, obj):
|
||
print(json.dumps(obj, indent=2))
|
||
|
||
# ------------------------------ workflows ------------------------------ #
|
||
def _workloads_menu(self):
|
||
workloads = self.client.list_workloads()
|
||
if not workloads:
|
||
print("No workloads found.")
|
||
return
|
||
lookup = {f"{w['name']} ({w['id'][:8]})": w for w in workloads}
|
||
choice = self._prompt("Select a workload:", list(lookup.keys()) + ["Back"])
|
||
if choice == "Back":
|
||
return
|
||
self._workload_actions(lookup[choice])
|
||
|
||
def _workload_actions(self, workload: Dict):
|
||
while True:
|
||
action = self._prompt(f"Workload {workload['name']} ›", self.ACTIONS_WORKLOAD)
|
||
if action == "Back":
|
||
return
|
||
if action == "Details":
|
||
details = self.client.get_workload(workload['id'])
|
||
self._pretty(details)
|
||
elif action in {"Start", "Stop", "Restart"}:
|
||
result = self.client.workload_action(workload['id'], action.lower())
|
||
self._pretty(result)
|
||
elif action == "Delete":
|
||
confirm = self._prompt("Confirm delete?", ["Yes", "No"])
|
||
if confirm == "Yes":
|
||
res = self.client._req("delete", f"/workloads/containers/{workload['id']}")
|
||
self._pretty(res)
|
||
return # workload gone; back to list
|
||
|
||
# ------------------------------ main loop ------------------------------ #
|
||
def run(self):
|
||
while True:
|
||
choice = self._prompt("Main ›", self.MAIN_MENU)
|
||
if choice == "Exit":
|
||
print("Goodbye!")
|
||
sys.exit(0)
|
||
if choice == "Workloads":
|
||
self._workloads_menu()
|
||
else:
|
||
print("Feature not implemented yet. 👷♂️")
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Entrypoint #
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def main() -> None:
|
||
"""Parse CLI args and start interactive session."""
|
||
parser = argparse.ArgumentParser(description="xCloudify interactive CLI")
|
||
parser.add_argument("--api-key", required=True, help="API key for authentication")
|
||
parser.add_argument("--base-url", default="http://localhost:5000/api", help="Base API URL (default: %(default)s)")
|
||
args = parser.parse_args()
|
||
|
||
client = ApiClient(base_url=args.base_url, api_key=args.api_key)
|
||
cli = XCloudifyCLI(client)
|
||
cli.run()
|
||
|
||
if __name__ == "__main__":
|
||
main()
|