Adds infrahubctl info command

Author: minitriga

changelog/109.added.md

@@ -0,0 +1 @@
+Adds `infrahubctl info` command to display information of the connectivity status of the SDK.

infrahub_sdk/ctl/cli_commands.py

@@ -4,6 +4,7 @@
 import functools
 import importlib
 import logging
+import platform
 import sys
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable
@@ -12,7 +13,11 @@
 import typer
 import ujson
 from rich.console import Console
+from rich.layout import Layout
 from rich.logging import RichHandler
+from rich.panel import Panel
+from rich.pretty import Pretty
+from rich.table import Table
 from rich.traceback import Traceback
 
 from .. import __version__ as sdk_version
@@ -34,6 +39,7 @@
 from ..ctl.utils import (
     catch_exception,
     execute_graphql_query,
+    get_user_permissions,
     load_yamlfile_from_disk_and_exit,
     parse_cli_vars,
 )
@@ -392,11 +398,177 @@ def protocols(
 
 @app.command(name="version")
 @catch_exception(console=console)
-def version(_: str = CONFIG_PARAM) -> None:
-    """Display the version of Infrahub and the version of the Python SDK in use."""
+def version() -> None:
+    """Display the version of Python and the version of the Python SDK in use."""
 
-    client = initialize_client_sync()
-    response = client.execute_graphql(query="query { InfrahubInfo { version }}")
+    console.print(f"Python: v{platform.python_version()}\nPython SDK: v{sdk_version}")
 
-    infrahub_version = response["InfrahubInfo"]["version"]
-    console.print(f"Infrahub: v{infrahub_version}\nPython SDK: v{sdk_version}")
+
+@app.command(name="info")
+@catch_exception(console=console)
+def info(detail: bool = typer.Option(False, help="Display detailed information."), _: str = CONFIG_PARAM) -> None:  # noqa: PLR0915
+    """Display the status of the Python SDK."""
+
+    query = """
+query GET_PROFILE_DETAILS {
+  AccountProfile {
+    id
+    display_label
+    account_type {
+      value
+      __typename
+      updated_at
+    }
+    status {
+      label
+      value
+      updated_at
+      __typename
+    }
+    description {
+      value
+      updated_at
+      __typename
+    }
+    label {
+      value
+      updated_at
+      __typename
+    }
+    member_of_groups {
+      count
+      edges {
+        node {
+          display_label
+          group_type {
+            value
+          }
+          ... on CoreAccountGroup {
+            id
+            roles {
+              count
+              edges {
+                node {
+                  permissions {
+                    count
+                    edges {
+                      node {
+                        display_label
+                        identifier {
+                          value
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }
+            display_label
+          }
+        }
+      }
+    }
+    __typename
+    name {
+      value
+      updated_at
+      __typename
+    }
+  }
+}
+"""
+
+    info: dict[str, Any] = {
+        "error": None,
+        "status": ":x:",
+        "infrahub_version": "N/A",
+        "user_info": None,
+        "groups": {},
+    }
+    try:
+        client = initialize_client_sync()
+        version_response = client.execute_graphql(query="query { InfrahubInfo { version }}")
+        info["infrahub_version"] = version_response["InfrahubInfo"]["version"]
+        info["user_info"] = client.execute_graphql(query=query)
+        info["status"] = ":white_heavy_check_mark:"
+        info["groups"] = get_user_permissions(info["user_info"]["AccountProfile"]["member_of_groups"]["edges"])
+    except Exception as e:
+        info["error"] = f"{e!s} ({e.__class__.__name__})"
+
+    if detail:
+        layout = Layout()
+
+        # Layout structure
+        new_console = Console(height=45)
+        layout = Layout()
+        layout.split_column(
+            Layout(name="body", ratio=1),
+        )
+        layout["body"].split_row(
+            Layout(name="left"),
+            Layout(name="right"),
+        )
+
+        layout["left"].split_column(
+            Layout(name="connection_status", size=7),
+            Layout(name="client_info", ratio=1),
+        )
+
+        layout["right"].split_column(
+            Layout(name="version_info", size=7),
+            Layout(name="infrahub_info", ratio=1),
+        )
+
+        # Connection status panel
+        connection_status = Table(show_header=False, box=None)
+        connection_status.add_row("Server Address:", client.config.address)
+        connection_status.add_row("Status:", info["status"])
+        if info["error"]:
+            connection_status.add_row("Error Reason:", info["error"])
+        layout["connection_status"].update(Panel(connection_status, title="Connection Status"))
+
+        # Version information panel
+        version_info = Table(show_header=False, box=None)
+        version_info.add_row("Python Version:", platform.python_version())
+        version_info.add_row("Infrahub Version", info["infrahub_version"])
+        version_info.add_row("Infrahub SDK:", sdk_version)
+        layout["version_info"].update(Panel(version_info, title="Version Information"))
+
+        # SDK client configuration panel
+        pretty_model = Pretty(client.config.dict(), expand_all=True)
+        layout["client_info"].update(Panel(pretty_model, title="Client Info"))
+
+        # Infrahub information planel
+        infrahub_info = Table(show_header=False, box=None)
+        if info["user_info"]:
+            infrahub_info.add_row("User:", info["user_info"]["AccountProfile"]["display_label"])
+            infrahub_info.add_row("Description:", info["user_info"]["AccountProfile"]["description"]["value"])
+            infrahub_info.add_row("Status:", info["user_info"]["AccountProfile"]["status"]["label"])
+            infrahub_info.add_row(
+                "Number of Groups:", str(info["user_info"]["AccountProfile"]["member_of_groups"]["count"])
+            )
+
+            if groups := info["groups"]:
+                infrahub_info.add_row("Groups:", "")
+                for k, v in info["groups"].items():
+                    infrahub_info.add_row("", k, v)
+
+        layout["infrahub_info"].update(Panel(infrahub_info, title="Infrahub Info"))
+
+        new_console.print(layout)
+    else:
+        # Simple output
+        table = Table(show_header=False, box=None)
+        table.add_row("Address:", client.config.address)
+        table.add_row("Connection Status", info["status"])
+        if info["error"]:
+            table.add_row("Connection Error:", info["error"])
+
+        table.add_row("Python Version:", platform.python_version())
+        table.add_row("SDK Version:", sdk_version)
+        table.add_row("Infrahub Version:", info["infrahub_version"])
+        if groups := info["groups"]:
+            table.add_row("Groups:")
+            for k, v in groups.items():
+                table.add_row("", k, v)
+        console.print(table)

infrahub_sdk/ctl/utils.py

@@ -214,3 +214,21 @@ def load_yamlfile_from_disk_and_exit(
         raise typer.Exit(1)
 
     return data_files
+
+
+def get_user_permissions(data: list[dict]) -> dict:
+    groups = {}
+    for group in data:
+        group_name = group["node"]["display_label"]
+        permissions = []
+
+        roles = group["node"].get("roles", {}).get("edges", [])
+        for role in roles:
+            role_permissions = role["node"].get("permissions", {}).get("edges", [])
+            for permission in role_permissions:
+                permissions.append(permission["node"]["identifier"]["value"])
+
+        permissions_str = ", ".join(permissions)
+        groups[group_name] = permissions_str
+
+    return groups