added base code for gnmi api client

gnmi_api_client/__init__.py

@@ -0,0 +1,3 @@
+from gnmi_api_client.client import GnmiApiClient, main
+
+__all__ = ["GnmiApiClient", "main"]

gnmi_api_client/client.py

@@ -0,0 +1,72 @@
+import argparse
+import shlex
+import sys
+
+from gnmi_api_client.input_formatter import ShowCliToGnmiPathConverter
+from gnmi_api_client.output_formatter import (
+    DummyTabularFormatter,
+    EXAMPLE_INTERFACES_STATUS_JSON,
+    EXAMPLE_INTERFACES_STATUS_PATH,
+)
+
+
+DEFAULT_HOST = "localhost"
+DEFAULT_PORT = 50052
+
+
+class GnmiApiClient:
+    def __init__(self, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, is_test_flag: bool = True):
+        self.host = host
+        self.port = port
+        self.is_test_flag = is_test_flag
+        self._output_formatter = DummyTabularFormatter()
+
+    def format_input(self, show_cli_str: str) -> str:
+        tokens = shlex.split(show_cli_str)
+        return ShowCliToGnmiPathConverter(tokens).convert()
+
+    def get_gnmi_result(self, gnmi_path: str) -> dict:
+        if self.is_test_flag:
+            if gnmi_path != EXAMPLE_INTERFACES_STATUS_PATH:
+                raise NotImplementedError(
+                    f"test-mode stub only supports {EXAMPLE_INTERFACES_STATUS_PATH}, got {gnmi_path}"
+                )
+            return EXAMPLE_INTERFACES_STATUS_JSON
+
+        # Real gNMI transport against f"{self.host}:{self.port}" is not wired up yet.
+        raise NotImplementedError(
+            f"real gNMI client against {self.host}:{self.port} is not implemented yet"
+        )
+
+    def format_output(self, gnmi_path: str, gnmi_json: dict) -> str:
+        return self._output_formatter.format(gnmi_path, gnmi_json)
+
+    def run(self, show_cli_str: str) -> str:
+        gnmi_path = self.format_input(show_cli_str)
+        gnmi_json = self.get_gnmi_result(gnmi_path)
+        return self.format_output(gnmi_path, gnmi_json)
+
+
+def main(argv=None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Convert a SONiC show CLI command into a gNMI SHOW path, fetch the result, and format it into CLI tabular format."
+    )
+    parser.add_argument("command", help="show CLI command string, e.g. 'show interfaces status'")
+    parser.add_argument("--host", default=DEFAULT_HOST, help="gNMI server host (default: localhost)")
+    parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="gNMI server port (default: 50052)")
+    parser.add_argument(
+        "--no-test-flag",
+        dest="is_test_flag",
+        action="store_false",
+        help="disable the hard-coded test fixture and attempt a real gNMI connection",
+    )
+    parser.set_defaults(is_test_flag=True)
+    args = parser.parse_args(argv)
+
+    client = GnmiApiClient(host=args.host, port=args.port, is_test_flag=args.is_test_flag)
+    print(client.run(args.command))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

gnmi_api_client/input_formatter.py

@@ -0,0 +1,76 @@
+NO_COMMAND_ERROR = "No command provided for conversion"
+EMPTY_COMMAND_ERROR = "Empty command"
+INVALID_COMMAND_ERROR = "Command must start with 'show'"
+NO_PATH_ERROR = "No path segments after 'show'"
+SHORT_OPTION_ERROR = "Short options are not supported"
+INVALID_OPTION_TOKEN_ERROR = "Invalid option token: --"
+INVALID_LONG_OPTION_ERROR = "Invalid long option '--'"
+
+
+class OptionException(Exception):
+    pass
+
+
+def has_special_char(text: str) -> bool:
+    return "=" in text or "]" in text or "[" in text
+
+
+def escape_gnmi(text: str) -> str:
+    # Escape only '/' → '\/'
+    return text.replace("/", r"\/")
+
+
+class ShowCliToGnmiPathConverter:
+    def __init__(self, tokens):
+        self.tokens = tokens
+
+    def parseLongOption(self, token: str):
+        # --flag         -> ('flag', 'True')
+        # --key=value    -> ('key',  'value')
+        if token == "--":
+            raise OptionException(INVALID_OPTION_TOKEN_ERROR)
+
+        body = token[2:]
+        if not body:
+            raise OptionException(INVALID_LONG_OPTION_ERROR)
+
+        if "=" in body:
+            name, value = body.split("=", 1)
+            if not name:
+                raise OptionException("Invalid long option: missing name before '='")
+            if has_special_char(name) or has_special_char(value):
+                raise OptionException("Invalid long option: key/value cannot contain =,[,]")
+            return name, escape_gnmi(value)
+
+        return body, "True"
+
+    def convert(self) -> str:
+        tokens = self.tokens
+        if not tokens:
+            raise OptionException(EMPTY_COMMAND_ERROR)
+        if tokens[0].lower() != "show":
+            raise OptionException(INVALID_COMMAND_ERROR)
+
+        tokens = tokens[1:]  # drop 'show'
+        out = ["SHOW"]
+
+        for tok in tokens:
+            if tok.startswith("-") and not tok.startswith("--"):
+                raise OptionException(f"{SHORT_OPTION_ERROR}: '{tok}'")
+
+            if tok.startswith("--"):
+                if len(out) == 1:
+                    raise ValueError("Option before first path segment")
+                key, val = self.parseLongOption(tok)
+                out.append(f"[{key}={val}]")
+                continue
+
+            if has_special_char(tok):
+                raise ValueError("Invalid characters inside of non option")
+
+            out.append("/")
+            out.append(escape_gnmi(tok))
+
+        if len(out) == 1:
+            raise OptionException(NO_PATH_ERROR)
+        return "".join(out)

gnmi_api_client/output_formatter.py

@@ -0,0 +1,45 @@
+from tabulate import tabulate
+
+
+EXAMPLE_INTERFACES_STATUS_PATH = "SHOW/interfaces/status"
+
+EXAMPLE_INTERFACES_STATUS_JSON = {
+    "interfaces": [
+        {
+            "Interface": "Ethernet0",
+            "Lanes": "0,1,2,3",
+            "Speed": "100G",
+            "MTU": "9100",
+            "FEC": "rs",
+            "Alias": "etp1",
+            "Admin": "up",
+            "Oper": "up",
+        },
+        {
+            "Interface": "Ethernet4",
+            "Lanes": "4,5,6,7",
+            "Speed": "100G",
+            "MTU": "9100",
+            "FEC": "rs",
+            "Alias": "etp2",
+            "Admin": "down",
+            "Oper": "down",
+        },
+    ]
+}
+
+_INTERFACES_STATUS_HEADERS = [
+    "Interface", "Lanes", "Speed", "MTU", "FEC", "Alias", "Admin", "Oper",
+]
+
+
+class DummyTabularFormatter:
+    def format(self, gnmi_path: str, gnmi_json: dict) -> str:
+        if gnmi_path != EXAMPLE_INTERFACES_STATUS_PATH:
+            raise ValueError(f"no dummy formatter for path {gnmi_path}")
+
+        rows = [
+            [entry.get(col, "") for col in _INTERFACES_STATUS_HEADERS]
+            for entry in gnmi_json.get("interfaces", [])
+        ]
+        return tabulate(rows, headers=_INTERFACES_STATUS_HEADERS, tablefmt="simple")

setup.py

@@ -58,6 +58,7 @@
         'crm',
         'debug',
         'generic_config_updater',
+        'gnmi_api_client',
         'dump',
         'dump.plugins',
         'pfcwd',