From: Vasek Sraier <git@vakabus.cz>
Date: Wed, 17 Aug 2022 08:44:26 +0000 (+0200)
Subject: kresctl: argument parsing infrastracture and init of tab completion
X-Git-Tag: v6.0.0a1~25^2~5
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=644f09df4f196dabdd8450b6a7d109ed5bd944eb;p=thirdparty%2Fknot-resolver.git

kresctl: argument parsing infrastracture and init of tab completion
---

diff --git a/manager/knot_resolver_manager/cli/__init__.py b/manager/knot_resolver_manager/cli/__init__.py
index d5fdd10f2..3355a3334 100644
--- a/manager/knot_resolver_manager/cli/__init__.py
+++ b/manager/knot_resolver_manager/cli/__init__.py
@@ -1,26 +1,124 @@
-from typing import TYPE_CHECKING, cast
+import argparse
+import importlib
+import pkgutil
+from abc import ABC, abstractmethod
+from typing import List, Optional, Tuple, Type, TypeVar, cast
 
-from typing_extensions import Literal
+from typing_extensions import Protocol
 
-from knot_resolver_manager.utils.requests import request
 
-if TYPE_CHECKING:
-    from knot_resolver_manager.cli.__main__ import Args, ConfigArgs
+class TopLevelArgs:
+    def __init__(self, ns: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
+        self.socket: str = ns.socket
+        self.command: Type["Command"] = ns.first_level_command
+        self.parser = parser
+        self.namespace = ns
 
 
-def config(args: "Args") -> None:
-    cfg: "ConfigArgs" = cast("ConfigArgs", args.command)
+class Command(ABC):
+    @staticmethod
+    @abstractmethod
+    def register_args_subparser(
+        parser: "argparse._SubParsersAction[argparse.ArgumentParser]",
+    ) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
+        raise NotImplementedError()
 
-    if not cfg.path.startswith("/"):
-        cfg.path = "/" + cfg.path
+    @abstractmethod
+    def __init__(self, ns: argparse.Namespace) -> None:
+        super().__init__()
 
-    method: Literal["GET", "POST"] = "GET" if cfg.replacement_value is None else "POST"
-    url = f"{args.socket}/v1/config{cfg.path}"
-    response = request(method, url, cfg.replacement_value)
-    print(response)
+    @abstractmethod
+    def run(self, args: TopLevelArgs) -> None:
+        raise NotImplementedError()
 
 
-def stop(args: "Args") -> None:
-    url = f"{args.socket}/stop"
-    response = request("POST", url)
-    print(response)
+_registered_commands: List[Type[Command]] = []
+
+
+T = TypeVar("T", bound=Type[Command])
+
+
+def register_command(cls: T) -> T:
+    _registered_commands.append(cls)
+    return cls
+
+
+def install_subcommand_parsers(arg: argparse.ArgumentParser) -> None:
+    subparsers = arg.add_subparsers()
+    for subcommand in _registered_commands:
+        parser, tp = subcommand.register_args_subparser(subparsers)
+        parser.set_defaults(first_level_command=tp)
+
+
+def create_main_arg_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser("kresctl", description="CLI for controlling Knot Resolver")
+    parser.add_argument(
+        "-s",
+        "--socket",
+        action="store",
+        type=str,
+        help="manager API listen address",
+        default="http+unix://%2Fvar%2Frun%2Fknot-resolver%2Fmanager.sock",  # FIXME
+        nargs=1,
+        required=False,
+    )
+    return parser
+
+
+class SubcommandProtocol(Protocol):
+    @staticmethod
+    def register_command(c: T) -> T:
+        raise NotImplementedError()
+
+
+def register_subcommand(
+    name: str, help: Optional[str] = None  # pylint: disable=redefined-builtin
+) -> SubcommandProtocol:
+    class Subcommand(Command):
+        subcommands: List[Type[Command]] = []
+
+        def __init__(self, ns: argparse.Namespace) -> None:
+            super().__init__(ns)
+            self.subcommand: Type[Command] = ns.subcommand
+
+        @staticmethod
+        def register_args_subparser(
+            parser: "argparse._SubParsersAction[argparse.ArgumentParser]",
+        ) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
+            subcmd = parser.add_parser(name, help=help)
+            subparsers = subcmd.add_subparsers()
+            for cmd in Subcommand.subcommands:
+                p, tp = cmd.register_args_subparser(subparsers)
+                p.set_defaults(subcommand=tp)
+
+            return subcmd, Subcommand
+
+        @staticmethod
+        def register_command(c: T) -> T:
+            Subcommand.subcommands.append(c)
+            return c
+
+        def run(self, args: TopLevelArgs) -> None:
+            cmd = self.subcommand(args.namespace)
+            cmd.run(args)
+
+    register_command(Subcommand)
+    return cast(SubcommandProtocol, Subcommand)
+
+
+def autoimport_commands() -> None:
+    for _loader, module_name, _is_pkg in pkgutil.walk_packages(
+        (f"{s}/cmd" for s in __path__), prefix="knot_resolver_manager.cli.cmd."
+    ):
+        importlib.import_module(module_name)
+
+
+def main() -> None:
+    autoimport_commands()
+    parser = create_main_arg_parser()
+    install_subcommand_parsers(parser)
+    ns = parser.parse_args()
+
+    toplevel = TopLevelArgs(ns, parser)
+    second = toplevel.command(ns)
+    second.run(toplevel)
diff --git a/manager/knot_resolver_manager/cli/__main__.py b/manager/knot_resolver_manager/cli/__main__.py
index b6ef92b66..56d61c984 100644
--- a/manager/knot_resolver_manager/cli/__main__.py
+++ b/manager/knot_resolver_manager/cli/__main__.py
@@ -1,80 +1,4 @@
-import argparse
-from abc import ABC
-from typing import Optional
-
-from knot_resolver_manager.cli import config, stop
-from knot_resolver_manager.compat.dataclasses import dataclass
-
-
-class Cmd(ABC):
-    def __init__(self, ns: argparse.Namespace) -> None:
-        pass
-
-    def run(self, args: "Args") -> None:
-        raise NotImplementedError()
-
-
-class ConfigArgs(Cmd):
-    def __init__(self, ns: argparse.Namespace) -> None:
-        super().__init__(ns)
-        self.path: str = str(ns.path)
-        self.replacement_value: Optional[str] = ns.new_value
-        self.delete: bool = ns.delete
-        self.stdin: bool = ns.stdin
-
-    def run(self, args: "Args") -> None:
-        config(args)
-
-
-class StopArgs(Cmd):
-    def run(self, args: "Args") -> None:
-        stop(args)
-
-
-@dataclass
-class Args:
-    socket: str
-    command: Cmd  # union in the future
-
-
-def parse_args() -> Args:
-    # pylint: disable=redefined-outer-name
-
-    parser = argparse.ArgumentParser("kresctl", description="CLI for controlling Knot Resolver")
-    parser.add_argument(
-        "-s",
-        "--socket",
-        action="store",
-        type=str,
-        help="manager API listen address",
-        default="http+unix://%2Fvar%2Frun%2Fknot-resolver%2Fmanager.sock",
-        nargs=1,
-        required=True,
-    )
-    subparsers = parser.add_subparsers()
-
-    config = subparsers.add_parser(
-        "config", help="dynamically change configuration of a running resolver", aliases=["c", "conf"]
-    )
-    config.add_argument("path", type=str, help="which part of config should we work with")
-    config.add_argument(
-        "new_value",
-        type=str,
-        nargs="?",
-        help="optional, what value should we set for the given path (JSON)",
-        default=None,
-    )
-    config.add_argument("-d", "--delete", action="store_true", help="delete part of the config tree", default=False)
-    config.add_argument("--stdin", help="read new config value on stdin", action="store_true", default=False)
-    config.set_defaults(command_type=ConfigArgs)
-
-    stop = subparsers.add_parser("stop", help="shutdown everything")
-    stop.set_defaults(command_type=StopArgs)
-
-    ns = parser.parse_args()
-    return Args(socket=ns.socket[0], command=ns.command_type(ns))  # type: ignore[call-arg]
-
+from knot_resolver_manager.cli import main
 
 if __name__ == "__main__":
-    _args = parse_args()
-    _args.command.run(_args)
+    main()
diff --git a/manager/knot_resolver_manager/cli/cmd/completion.py b/manager/knot_resolver_manager/cli/cmd/completion.py
new file mode 100644
index 000000000..f10e4858a
--- /dev/null
+++ b/manager/knot_resolver_manager/cli/cmd/completion.py
@@ -0,0 +1,59 @@
+import argparse
+from typing import Dict, Optional, Tuple, Type
+
+from knot_resolver_manager.cli import Command, TopLevelArgs, register_subcommand
+
+
+def _list_subcommands(parser: argparse.ArgumentParser) -> Dict[str, Optional[str]]:
+    try:
+        result: Dict[str, Optional[str]] = {}
+        for action in parser._subparsers._actions:  # type: ignore
+            if isinstance(action, argparse._SubParsersAction):
+                for subact in action._get_subactions():
+                    name = subact.dest
+                    help = subact.help
+                    result[name] = help
+        return result
+    except Exception:
+        # if it fails, abort
+        return {}
+
+
+Completions = register_subcommand("completion", help="shell completions")
+
+
+@Completions.register_command
+class FishCompletion(Command):
+    def __init__(self, ns: argparse.Namespace) -> None:
+        super().__init__(ns)
+
+    @staticmethod
+    def register_args_subparser(
+        parser: "argparse._SubParsersAction[argparse.ArgumentParser]",
+    ) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
+        fcpl = parser.add_parser("fish", help="completion for fish")
+        return fcpl, FishCompletion
+
+    def run(self, args: TopLevelArgs) -> None:
+        for cmd, help in _list_subcommands(args.parser).items():
+            print(f"complete -c kresctl -a '{cmd}'", end="")
+            if help is not None:
+                print(f" -d '{help}'")
+            else:
+                print()
+
+
+@Completions.register_command
+class BashCompletion(Command):
+    def __init__(self, ns: argparse.Namespace) -> None:
+        super().__init__(ns)
+
+    @staticmethod
+    def register_args_subparser(
+        parser: "argparse._SubParsersAction[argparse.ArgumentParser]",
+    ) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
+        fcpl = parser.add_parser("bash", help="completion for bash")
+        return fcpl, BashCompletion
+
+    def run(self, args: TopLevelArgs) -> None:
+        raise NotImplementedError
diff --git a/manager/knot_resolver_manager/cli/cmd/config.py b/manager/knot_resolver_manager/cli/cmd/config.py
new file mode 100644
index 000000000..029b6f8af
--- /dev/null
+++ b/manager/knot_resolver_manager/cli/cmd/config.py
@@ -0,0 +1,46 @@
+import argparse
+from typing import Optional, Tuple, Type
+
+from typing_extensions import Literal
+
+from knot_resolver_manager.cli import Command, TopLevelArgs, register_command
+from knot_resolver_manager.utils.requests import request
+
+
+@register_command
+class ConfigCmd(Command):
+    def __init__(self, ns: argparse.Namespace) -> None:
+        super().__init__(ns)
+        self.path: str = str(ns.path)
+        self.replacement_value: Optional[str] = ns.new_value
+        self.delete: bool = ns.delete
+        self.stdin: bool = ns.stdin
+
+    @staticmethod
+    def register_args_subparser(
+        parser: "argparse._SubParsersAction[argparse.ArgumentParser]",
+    ) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
+        config = parser.add_parser(
+            "config", help="dynamically change configuration of a running resolver", aliases=["c", "conf"]
+        )
+        config.add_argument("path", type=str, help="which part of config should we work with")
+        config.add_argument(
+            "new_value",
+            type=str,
+            nargs="?",
+            help="optional, what value should we set for the given path (JSON)",
+            default=None,
+        )
+        config.add_argument("-d", "--delete", action="store_true", help="delete part of the config tree", default=False)
+        config.add_argument("--stdin", help="read new config value on stdin", action="store_true", default=False)
+
+        return config, ConfigCmd
+
+    def run(self, args: TopLevelArgs) -> None:
+        if not self.path.startswith("/"):
+            self.path = "/" + self.path
+
+        method: Literal["GET", "POST"] = "GET" if self.replacement_value is None else "POST"
+        url = f"{args.socket}/v1/config{self.path}"
+        response = request(method, url, self.replacement_value)
+        print(response)
diff --git a/manager/knot_resolver_manager/cli/cmd/stop.py b/manager/knot_resolver_manager/cli/cmd/stop.py
new file mode 100644
index 000000000..a9d899459
--- /dev/null
+++ b/manager/knot_resolver_manager/cli/cmd/stop.py
@@ -0,0 +1,23 @@
+import argparse
+from typing import Tuple, Type
+
+from knot_resolver_manager.cli import Command, TopLevelArgs, register_command
+from knot_resolver_manager.utils.requests import request
+
+
+@register_command
+class StopCmd(Command):
+    def __init__(self, ns: argparse.Namespace) -> None:
+        super().__init__(ns)
+
+    def run(self, args: TopLevelArgs) -> None:
+        url = f"{args.socket}/stop"
+        response = request("POST", url)
+        print(response)
+
+    @staticmethod
+    def register_args_subparser(
+        parser: "argparse._SubParsersAction[argparse.ArgumentParser]",
+    ) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
+        stop = parser.add_parser("stop", help="shutdown everything")
+        return stop, StopCmd
diff --git a/manager/pyproject.toml b/manager/pyproject.toml
index 26d65cb40..b9b66a656 100644
--- a/manager/pyproject.toml
+++ b/manager/pyproject.toml
@@ -54,7 +54,7 @@ format = { shell = "black knot_resolver_manager/ tests/ scripts/ build.py; isort
 fixdeps = { shell = "poetry install; npm install; npm update", help = "Install/update dependencies according to configuration files"}
 commit = { shell = "scripts/commit", help = "Invoke every single check before commiting" }
 container = { cmd = "scripts/container.py", help = "Manage containers" }
-client = { script = "knot_resolver_manager.client.__main__:main", help="Run Managers API client CLI" }
+kresctl = { script = "knot_resolver_manager.cli:main", help="run kresctl" }
 clean = """
   rm -rf .coverage
          .mypy_cache
diff --git a/manager/setup.py b/manager/setup.py
index 96ae75970..6377e5e11 100644
--- a/manager/setup.py
+++ b/manager/setup.py
@@ -4,6 +4,7 @@ from setuptools import setup
 packages = \
 ['knot_resolver_manager',
  'knot_resolver_manager.cli',
+ 'knot_resolver_manager.cli.cmd',
  'knot_resolver_manager.compat',
  'knot_resolver_manager.datamodel',
  'knot_resolver_manager.datamodel.types',