-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)
-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()
--- /dev/null
+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
--- /dev/null
+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)
--- /dev/null
+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
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
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',
projects / thirdparty / knot-resolver.git / commitdiff
