from knot_resolver_manager.utils.etag import structural_etag
from knot_resolver_manager.utils.functional import Result
from knot_resolver_manager.utils.modeling.exceptions import DataParsingError, DataValidationError
-from knot_resolver_manager.utils.modeling.parsing import DataFormat, parse_yaml
+from knot_resolver_manager.utils.modeling.parsing import DataFormat, try_to_parse
from knot_resolver_manager.utils.modeling.query import query
from knot_resolver_manager.utils.modeling.types import NoneType
from knot_resolver_manager.utils.systemd_notify import systemd_notify
else:
try:
data = await readfile(self._config_path)
- config = KresConfig(parse_yaml(data))
+ config = KresConfig(try_to_parse(data))
await self.config_store.update(config)
logger.info("Configuration file successfully reloaded")
except FileNotFoundError:
)
else:
logger.info("Loading initial configuration from %s", config)
- config = parse_yaml(await readfile(config))
+ config = try_to_parse(await readfile(config))
# validate the initial configuration
assert isinstance(config, dict)
def try_to_parse(data: str) -> Any:
- """Attempt to parse the data as a YAML or JSON string."""
+ """Attempt to parse the data as a JSON or YAML string."""
+
try:
- return parse_yaml(data)
- except yaml.YAMLError as ye:
+ return parse_json(data)
+ except json.JSONDecodeError as je:
try:
- return parse_json(data)
- except json.JSONDecodeError as je:
- raise DataParsingError(f"failed to parse data, YAML: {ye}, JSON: {je}")
+ return parse_yaml(data)
+ except yaml.YAMLError as ye:
+ raise DataParsingError(f"failed to parse data, JSON: {je}, YAML: {ye}")