This commit is contained in:
vy.boyko 2025-10-26 22:02:25 +03:00
parent 83ff9af4d6
commit 969a84f30f

View File

@ -44,7 +44,8 @@ class K8sTool:
"""Run the main application loop.""" """Run the main application loop."""
console.print(Panel.fit( console.print(Panel.fit(
"[bold cyan]K8s Tool[/bold cyan]\n" "[bold cyan]K8s Tool[/bold cyan]\n"
"[dim]Interactive kubectl helper[/dim]", "[dim]Interactive kubectl helper[/dim]\n"
"[dim]Tip: Press Ctrl+C to cancel and go back[/dim]",
border_style="cyan" border_style="cyan"
)) ))
@ -93,6 +94,10 @@ class K8sTool:
style=custom_style style=custom_style
).ask() ).ask()
if not action:
# User pressed Esc
return
if action == "Manage Contexts": if action == "Manage Contexts":
self._manage_contexts() self._manage_contexts()
elif action == "Select Namespace": elif action == "Select Namespace":
@ -274,12 +279,19 @@ class K8sTool:
default=False default=False
).ask() ).ask()
if has_token is None:
# User pressed Esc
return
token = None token = None
if has_token: if has_token:
token = questionary.password( token = questionary.password(
"Authentication token:", "Authentication token:",
style=custom_style style=custom_style
).ask() ).ask()
if token is None:
# User pressed Esc
return
# Optional: certificate # Optional: certificate
has_cert = questionary.confirm( has_cert = questionary.confirm(
@ -288,12 +300,19 @@ class K8sTool:
default=False default=False
).ask() ).ask()
if has_cert is None:
# User pressed Esc
return
cert = None cert = None
if has_cert: if has_cert:
cert = questionary.text( cert = questionary.text(
"Certificate authority data (base64 encoded):", "Certificate authority data (base64 encoded):",
style=custom_style style=custom_style
).ask() ).ask()
if not cert:
# User pressed Esc
return
# Confirm # Confirm
console.print("\n[bold]Review:[/bold]") console.print("\n[bold]Review:[/bold]")
@ -304,7 +323,8 @@ class K8sTool:
console.print(f" Token: [cyan]{'***' if token else 'None'}[/cyan]") console.print(f" Token: [cyan]{'***' if token else 'None'}[/cyan]")
console.print(f" Certificate: [cyan]{'***' if cert else 'None (insecure)'}[/cyan]") console.print(f" Certificate: [cyan]{'***' if cert else 'None (insecure)'}[/cyan]")
if questionary.confirm("\nAdd this context?", style=custom_style, default=False).ask(): confirm = questionary.confirm("\nAdd this context?", style=custom_style, default=False).ask()
if confirm:
self.k8s_client.add_context(context_name, cluster_name, user_name, server, cert, token) self.k8s_client.add_context(context_name, cluster_name, user_name, server, cert, token)
def _delete_context(self): def _delete_context(self):
@ -332,11 +352,13 @@ class K8sTool:
if not context_name or context_name == "Cancel": if not context_name or context_name == "Cancel":
return return
if questionary.confirm( confirm = questionary.confirm(
f"Are you sure you want to delete context '{context_name}'?", f"Are you sure you want to delete context '{context_name}'?",
style=custom_style, style=custom_style,
default=False default=False
).ask(): ).ask()
if confirm:
self.k8s_client.delete_context(context_name) self.k8s_client.delete_context(context_name)
def _manage_favorites(self): def _manage_favorites(self):
@ -384,7 +406,8 @@ class K8sTool:
self._remove_namespace_from_favorites() self._remove_namespace_from_favorites()
elif action == "Clear all favorites": elif action == "Clear all favorites":
if questionary.confirm("Are you sure you want to clear all favorites?", style=custom_style, default=False).ask(): confirm = questionary.confirm("Are you sure you want to clear all favorites?", style=custom_style, default=False).ask()
if confirm:
self.config.clear_favorites() self.config.clear_favorites()
def _view_favorites(self): def _view_favorites(self):
@ -466,7 +489,8 @@ class K8sTool:
self.k8s_client.display_deployments_table(deployments) self.k8s_client.display_deployments_table(deployments)
if questionary.confirm("View pods for a deployment?", style=custom_style, default=False).ask(): view_pods = questionary.confirm("View pods for a deployment?", style=custom_style, default=False).ask()
if view_pods:
dep_names = [dep['name'] for dep in deployments] dep_names = [dep['name'] for dep in deployments]
dep_name = questionary.select( dep_name = questionary.select(
"Select deployment:", "Select deployment:",
@ -653,7 +677,8 @@ class K8sTool:
self.k8s_client.display_configmaps_table(configmaps) self.k8s_client.display_configmaps_table(configmaps)
if questionary.confirm("View ConfigMap data?", style=custom_style, default=False).ask(): view_data = questionary.confirm("View ConfigMap data?", style=custom_style, default=False).ask()
if view_data:
cm_names = [cm['name'] for cm in configmaps] cm_names = [cm['name'] for cm in configmaps]
cm_name = questionary.select( cm_name = questionary.select(
"Select ConfigMap:", "Select ConfigMap:",
@ -784,7 +809,7 @@ class K8sTool:
console.print("[yellow]No pods found[/yellow]") console.print("[yellow]No pods found[/yellow]")
return return
pod_names = [f"{pod['name']} ({pod['status']})" for pod in pods] pod_names = [f"{pod['name']} (Status: {pod['status']}, Age: {pod['age']})" for pod in pods]
selected = questionary.select( selected = questionary.select(
"Select pod:", "Select pod:",
choices=pod_names, choices=pod_names,
@ -794,7 +819,7 @@ class K8sTool:
if not selected: if not selected:
return return
pod_name = selected.split(" (")[0] pod_name = selected.split(" (Status:")[0]
# Check if pod has multiple containers # Check if pod has multiple containers
containers = self.k8s_client.get_pod_containers(self.current_namespace, pod_name) containers = self.k8s_client.get_pod_containers(self.current_namespace, pod_name)
@ -806,6 +831,9 @@ class K8sTool:
choices=containers, choices=containers,
style=custom_style style=custom_style
).ask() ).ask()
if not container:
# User pressed Esc
return
elif len(containers) == 1: elif len(containers) == 1:
container = containers[0] container = containers[0]