Adds infrahubctl info command

Co-authored-by: Alex Gittings <[email protected]>

Author: dgarros

changelog/109.added.md

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

infrahub_sdk/client.py

@@ -46,13 +46,13 @@
 )
 from .object_store import ObjectStore, ObjectStoreSync
 from .protocols_base import CoreNode, CoreNodeSync
-from .queries import get_commit_update_mutation
+from .queries import QUERY_USER, get_commit_update_mutation
 from .query_groups import InfrahubGroupContext, InfrahubGroupContextSync
 from .schema import InfrahubSchema, InfrahubSchemaSync, NodeSchemaAPI
 from .store import NodeStore, NodeStoreSync
 from .timestamp import Timestamp
 from .types import AsyncRequester, HTTPMethod, SyncRequester
-from .utils import decode_json, is_valid_uuid
+from .utils import decode_json, get_user_permissions, is_valid_uuid
 
 if TYPE_CHECKING:
     from types import TracebackType
@@ -272,6 +272,22 @@ def _initialize(self) -> None:
         self._request_method: AsyncRequester = self.config.requester or self._default_request_method
         self.group_context = InfrahubGroupContext(self)
 
+    async def get_version(self) -> str:
+        """Return the Infrahub version."""
+        response = await self.execute_graphql(query="query { InfrahubInfo { version }}")
+        version = response.get("InfrahubInfo", {}).get("version", "")
+        return version
+
+    async def get_user(self) -> dict:
+        """Return user information"""
+        user_info = await self.execute_graphql(query=QUERY_USER)
+        return user_info
+
+    async def get_user_permissions(self) -> dict:
+        """Return user permissions"""
+        user_info = await self.get_user()
+        return get_user_permissions(user_info["AccountProfile"]["member_of_groups"]["edges"])
+
     @overload
     async def create(
         self,
@@ -1479,6 +1495,22 @@ def _initialize(self) -> None:
         self._request_method: SyncRequester = self.config.sync_requester or self._default_request_method
         self.group_context = InfrahubGroupContextSync(self)
 
+    def get_version(self) -> str:
+        """Return the Infrahub version."""
+        response = self.execute_graphql(query="query { InfrahubInfo { version }}")
+        version = response.get("InfrahubInfo", {}).get("version", "")
+        return version
+
+    def get_user(self) -> dict:
+        """Return user information"""
+        user_info = self.execute_graphql(query=QUERY_USER)
+        return user_info
+
+    def get_user_permissions(self) -> dict:
+        """Return user permissions"""
+        user_info = self.get_user()
+        return get_user_permissions(user_info["AccountProfile"]["member_of_groups"]["edges"])
+
     @overload
     def create(
         self,

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
@@ -392,11 +397,106 @@ 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: {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."""
+
+    info: dict[str, Any] = {
+        "error": None,
+        "status": ":x:",
+        "infrahub_version": "N/A",
+        "user_info": {},
+        "groups": {},
+    }
+    try:
+        client = initialize_client_sync()
+        info["infrahub_version"] = client.get_version()
+        info["user_info"] = client.get_user()
+        info["status"] = ":white_heavy_check_mark:"
+        info["groups"] = client.get_user_permissions()
+    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.model_dump(), 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 group, roles in groups.items():
+                    infrahub_info.add_row("", group, ", ".join(roles))
+
+        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 account := info["user_info"].get("AccountProfile"):
+            table.add_row("User:", account["display_label"])
+
+        console.print(table)

infrahub_sdk/queries.py

@@ -42,3 +42,72 @@ def get_commit_update_mutation(is_read_only: bool = False) -> str:
   }
 }
 """
+
+QUERY_USER = """
+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
+    }
+  }
+}
+"""

infrahub_sdk/utils.py

@@ -335,3 +335,20 @@ def write_to_file(path: Path, value: Any) -> bool:
     written = path.write_text(to_write)
 
     return written is not None
+
+
+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"])
+
+        groups[group_name] = permissions
+
+    return groups

tests/fixtures/account_profile.json

@@ -0,0 +1,79 @@
+{
+    "data": {
+        "AccountProfile": {
+            "id": "1816ebcd-cea7-3bf7-3fc9-c51282f03fe7",
+            "display_label": "Admin",
+            "account_type": {
+                "value": "User",
+                "__typename": "TextAttribute",
+                "updated_at": "2025-01-02T16:06:15.565985+00:00"
+            },
+            "status": {
+                "label": "Active",
+                "value": "active",
+                "updated_at": "2025-01-02T16:06:15.565985+00:00",
+                "__typename": "Dropdown"
+            },
+            "description": {
+                "value": null,
+                "updated_at": "2025-01-02T16:06:15.565985+00:00",
+                "__typename": "TextAttribute"
+            },
+            "label": {
+                "value": "Admin",
+                "updated_at": "2025-01-02T16:06:15.565985+00:00",
+                "__typename": "TextAttribute"
+            },
+            "member_of_groups": {
+                "count": 1,
+                "edges": [
+                    {
+                        "node": {
+                            "display_label": "Super Administrators",
+                            "group_type": {
+                                "value": "default"
+                            },
+                            "id": "1816ebce-1cbe-2e96-3fc3-c5124c324bac",
+                            "roles": {
+                                "count": 1,
+                                "edges": [
+                                    {
+                                        "node": {
+                                            "permissions": {
+                                                "count": 2,
+                                                "edges": [
+                                                    {
+                                                        "node": {
+                                                            "display_label": "super_admin 6",
+                                                            "identifier": {
+                                                                "value": "global:super_admin:allow_all"
+                                                            }
+                                                        }
+                                                    },
+                                                    {
+                                                        "node": {
+                                                            "display_label": "* * any 6",
+                                                            "identifier": {
+                                                                "value": "object:*:*:any:allow_all"
+                                                            }
+                                                        }
+                                                    }
+                                                ]
+                                            }
+                                        }
+                                    }
+                                ]
+                            }
+                        }
+                    }
+                ]
+            },
+            "__typename": "CoreAccount",
+            "name": {
+                "value": "admin",
+                "updated_at": "2025-01-02T16:06:15.565985+00:00",
+                "__typename": "TextAttribute"
+            }
+        }
+    }
+}

tests/unit/ctl/conftest.py

@@ -1,6 +1,8 @@
 import pytest
 from pytest_httpx import HTTPXMock
 
+from tests.unit.sdk.conftest import mock_query_infrahub_user, mock_query_infrahub_version  # noqa: F401
+
 
 @pytest.fixture
 async def mock_branches_list_query(httpx_mock: HTTPXMock) -> HTTPXMock: