import argparse
import json
+import sys
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple, Type
new_config = self.value_or_file
response = request(method, url, reformat(new_config, Formats.JSON) if new_config else None)
+
+ if response.status != 200:
+ print(response)
+ sys.exit(1)
+
print(f"status: {response.status}")
- if self.operation == Operations.GET and response.status == 200:
- if self.value_or_file:
- with open(self.value_or_file, "w") as f:
- f.write(reformat(response.body, self.format))
- print(f"saved to: {self.value_or_file}")
- else:
- print(reformat(response.body, self.format))
+ if self.operation == Operations.GET and self.value_or_file:
+ with open(self.value_or_file, "w") as f:
+ f.write(reformat(response.body, self.format))
+ print(f"saved to: {self.value_or_file}")
else:
- print(response.body)
+ print(reformat(response.body, self.format))
import argparse
+import sys
from typing import List, Optional, Tuple, Type
from knot_resolver_manager.cli.command import Command, CommandArgs, CompWords, register_command
from knot_resolver_manager.datamodel import KresConfig
from knot_resolver_manager.utils.modeling import try_to_parse
-from knot_resolver_manager.utils.modeling.exceptions import DataParsingError
+from knot_resolver_manager.utils.modeling.exceptions import DataParsingError, DataValidationError
@register_command
try:
parsed = try_to_parse(data)
- except DataParsingError as e:
+ lua = KresConfig(parsed).render_lua()
+ except (DataParsingError, DataValidationError) as e:
print(e)
- return
-
- lua = KresConfig(parsed).render_lua()
+ sys.exit(1)
if self.output_file:
with open(self.output_file, "w") as f:
import argparse
+import sys
from typing import List, Optional, Tuple, Type
from knot_resolver_manager.cli.command import Command, CommandArgs, CompWords, register_command
url = f"{args.socket}/metrics"
response = request("GET", url)
- if self.file and response.status == 200:
- with open(self.file, "w") as f:
- f.write(response.body)
+ if response.status == 200:
+ if self.file:
+ with open(self.file, "w") as f:
+ f.write(response.body)
+ else:
+ print(response.body)
else:
print(response)
+ sys.exit(1)
import argparse
+import sys
from typing import List, Tuple, Type
from knot_resolver_manager.cli.command import Command, CommandArgs, CompWords, register_command
url = f"{args.socket}/reload"
response = request("POST", url)
print(response)
+
+ if response.status != 200:
+ sys.exit(1)
import argparse
import json
+import sys
from typing import List, Optional, Tuple, Type
from knot_resolver_manager.cli.command import Command, CommandArgs, CompWords, parser_words, register_command
schema = response.body
else:
print(response)
- return
+ sys.exit(1)
else:
schema = json.dumps(KresConfig.json_schema(), indent=4)
import argparse
+import sys
from typing import List, Tuple, Type
from knot_resolver_manager.cli.command import Command, CommandArgs, CompWords, register_command
response = request("POST", url)
print(response)
+ if response.status != 200:
+ sys.exit(1)
+
@staticmethod
def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
return {}
import argparse
+import sys
from typing import List, Tuple, Type
from knot_resolver_manager.cli.command import Command, CommandArgs, CompWords, register_command
print("config is valid")
except (DataParsingError, DataValidationError) as e:
print(e)
+ sys.exit(1)