1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
import argparse
import json
import sys
from typing import List, Optional, Tuple, Type
from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
from knot_resolver.datamodel import kres_config_json_schema
from knot_resolver.utils.requests import request
@register_command
class SchemaCommand(Command):
def __init__(self, namespace: argparse.Namespace) -> None:
super().__init__(namespace)
self.live: bool = namespace.live
self.file: Optional[str] = namespace.file
@staticmethod
def register_args_subparser(
subparser: "argparse._SubParsersAction[argparse.ArgumentParser]",
) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
schema = subparser.add_parser(
"schema", help="Shows JSON-schema repersentation of the Knot Resolver's configuration."
)
schema.add_argument(
"-l",
"--live",
help="Get configuration JSON-schema from the running resolver. Requires connection to the management API.",
action="store_true",
default=False,
)
schema.add_argument("file", help="Optional, file where to export JSON-schema.", nargs="?", default=None)
return schema, SchemaCommand
@staticmethod
def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
return comp_get_words(args, parser._actions) # noqa: SLF001
def run(self, args: CommandArgs) -> None:
if self.live:
response = request(args.socket, "GET", "schema")
if response.status != 200:
print(response, file=sys.stderr)
sys.exit(1)
schema = response.body
else:
schema = json.dumps(kres_config_json_schema(), indent=4)
if self.file:
with open(self.file, "w") as f:
f.write(schema)
else:
print(schema)
|
