-import argparse
-import importlib
-import pkgutil
-from abc import ABC, abstractmethod
-from pathlib import Path
-from typing import List, Optional, Tuple, Type, TypeVar, cast
-from urllib.parse import quote
-
-from typing_extensions import Protocol
-
-
-class TopLevelArgs:
- def __init__(self, ns: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
- self.socket: str = ns.socket[0]
- if Path(self.socket).exists():
- self.socket = f'http+unix://{quote(self.socket, safe="")}/'
- if self.socket.endswith("/"):
- self.socket = self.socket[:-1]
-
- self.command: Type["Command"] = ns.first_level_command
- self.parser = parser
- self.namespace = ns
-
-
-class Command(ABC):
- @staticmethod
- @abstractmethod
- def register_args_subparser(
- parser: "argparse._SubParsersAction[argparse.ArgumentParser]",
- ) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
- raise NotImplementedError()
-
- @abstractmethod
- def __init__(self, ns: argparse.Namespace) -> None:
- super().__init__()
-
- @abstractmethod
- def run(self, args: TopLevelArgs) -> None:
- raise NotImplementedError()
-
-
-_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)
-from knot_resolver_manager.cli import main
+import argparse
+import importlib
+import os
+
+from knot_resolver_manager.cli.command import install_commands_parsers
+from knot_resolver_manager.cli.kresctl import Kresctl
+
+
+def autoimport_commands() -> None:
+ prefix = "knot_resolver_manager.cli.cmd."
+ for module_name in os.listdir(os.path.dirname(__file__) + "/cmd"):
+ if module_name[-3:] != ".py":
+ continue
+ importlib.import_module(f"{prefix}{module_name[:-3]}")
+
+
+def create_main_argument_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser("kresctl", description="Command-line interface for controlling Knot Resolver")
+ parser.add_argument(
+ "-i",
+ "--interactive",
+ action="store_true",
+ help="Interactive mode of kresctl utility",
+ default=False,
+ required=False,
+ )
+ parser.add_argument(
+ "-s",
+ "--socket",
+ action="store",
+ type=str,
+ help="Path to the Unix domain socket of the configuration API",
+ default=["http+unix://%2Fvar%2Frun%2Fknot-resolver%2Fmanager.sock"], # FIXME
+ nargs=1,
+ required=False,
+ )
+ return parser
+
+
+def main() -> None:
+ autoimport_commands()
+ parser = create_main_argument_parser()
+ install_commands_parsers(parser)
+ namespace = parser.parse_args()
+ kresctl = Kresctl(namespace, parser)
+
+ if namespace.interactive or len(vars(namespace)) == 2:
+ kresctl.interactive()
+ else:
+ kresctl.execute()
+
if __name__ == "__main__":
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
from typing_extensions import Literal
-from knot_resolver_manager.cli import Command, TopLevelArgs, register_command
+from knot_resolver_manager.cli.command import Command, CommandArgs, 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
+class ConfigCommand(Command):
+ def __init__(self, namespace: argparse.Namespace) -> None:
+ super().__init__(namespace)
+ self.path: str = str(namespace.path)
+ self.replacement_value: Optional[str] = namespace.new_value
+ self.delete: bool = namespace.delete
+ self.stdin: bool = namespace.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 = parser.add_parser("config", help="dynamically change configuration of a running resolver")
config.add_argument("path", type=str, help="which part of config should we work with")
config.add_argument(
"new_value",
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
+ return config, ConfigCommand
- def run(self, args: TopLevelArgs) -> None:
+ def run(self, args: CommandArgs) -> None:
if not self.path.startswith("/"):
self.path = "/" + self.path
--- /dev/null
+import argparse
+import sys
+from typing import Tuple, Type
+
+from knot_resolver_manager.cli.command import Command, CommandArgs, register_command
+
+
+@register_command
+class ExitCommand(Command):
+ def __init__(self, namespace: argparse.Namespace) -> None:
+ super().__init__(namespace)
+
+ def run(self, args: CommandArgs) -> None:
+ sys.exit()
+
+ @staticmethod
+ def register_args_subparser(
+ parser: "argparse._SubParsersAction[argparse.ArgumentParser]",
+ ) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
+ stop = parser.add_parser("exit", help="exit kresctl")
+ return stop, ExitCommand
import argparse
from typing import Tuple, Type
-from knot_resolver_manager.cli import Command, TopLevelArgs, register_command
+from knot_resolver_manager.cli.command import Command, CommandArgs, 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)
+class StopCommand(Command):
+ def __init__(self, namespace: argparse.Namespace) -> None:
+ super().__init__(namespace)
- def run(self, args: TopLevelArgs) -> None:
+ def run(self, args: CommandArgs) -> None:
url = f"{args.socket}/stop"
response = request("POST", url)
print(response)
parser: "argparse._SubParsersAction[argparse.ArgumentParser]",
) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
stop = parser.add_parser("stop", help="shutdown everything")
- return stop, StopCmd
+ return stop, StopCommand
--- /dev/null
+import argparse
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import List, Tuple, Type, TypeVar
+from urllib.parse import quote
+
+T = TypeVar("T", bound=Type["Command"])
+
+_registered_commands: List[Type["Command"]] = []
+
+
+def register_command(cls: T) -> T:
+ _registered_commands.append(cls)
+ return cls
+
+
+def install_commands_parsers(parser: argparse.ArgumentParser) -> None:
+ subparsers = parser.add_subparsers(help="command type")
+ for command in _registered_commands:
+ subparser, typ = command.register_args_subparser(subparsers)
+ subparser.set_defaults(command=typ)
+
+
+class CommandArgs:
+ def __init__(self, namespace: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
+ self.socket: str = namespace.socket[0]
+ if Path(self.socket).exists():
+ self.socket = f'http+unix://{quote(self.socket, safe="")}/'
+ if self.socket.endswith("/"):
+ self.socket = self.socket[:-1]
+
+ self.command: Type["Command"] = namespace.command
+ self.parser = parser
+ self.namespace = namespace
+
+
+class Command(ABC):
+ @staticmethod
+ @abstractmethod
+ def register_args_subparser(
+ parser: argparse.ArgumentParser,
+ ) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
+ raise NotImplementedError()
+
+ @abstractmethod
+ def __init__(self, namespace: argparse.Namespace) -> None:
+ super().__init__()
+
+ @abstractmethod
+ def run(self, args: CommandArgs) -> None:
+ raise NotImplementedError()
--- /dev/null
+import argparse
+
+from knot_resolver_manager.cli.command import CommandArgs
+
+
+class Kresctl:
+ def __init__(self, namespace: argparse.Namespace, parser: argparse.ArgumentParser, prompt: str = "kresctl") -> None:
+ self.path = None
+ self.prompt = prompt
+ self.namespace = namespace
+ self.parser = parser
+
+ def execute(self):
+ cmd_args = CommandArgs(self.namespace, self.parser)
+ command = self.namespace.command(self.namespace)
+ command.run(cmd_args)
+
+ def _prompt_format(self) -> str:
+ bolt = "\033[1m"
+ white = "\033[38;5;255m"
+ reset = "\033[0;0m"
+
+ if self.path:
+ prompt = f"{bolt}[{self.prompt} {white}{self.path}{reset}{bolt}]"
+ else:
+ prompt = f"{bolt}{self.prompt}"
+ return f"{prompt}> {reset}"
+
+ def interactive(self):
+ try:
+ while True:
+ pass
+ # TODO: not working yet
+ # cmd = input(f"{self._prompt_format()}")
+ # namespace = self.parser.parse_args(cmd.split(" "))
+ # namespace.interactive = True
+ # namespace.socket = self.namespace.socket
+ # self.namespace = namespace
+ # self.execute()
+ except KeyboardInterrupt:
+ pass
projects / thirdparty / knot-resolver.git / commitdiff
