manager: api: allow JSON only as MIME type

manager/knot_resolver_manager/server.py

@@ -33,7 +33,7 @@ from knot_resolver_manager.utils.async_utils import readfile
 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 parse, parse_yaml
+from knot_resolver_manager.utils.modeling.parsing import DataFormat, parse_yaml
 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
@@ -62,6 +62,20 @@ async def error_handler(request: web.Request, handler: Any) -> web.Response:
         return web.Response(text=f"request processing failed:\n{e}", status=HTTPStatus.INTERNAL_SERVER_ERROR)
 
 
+def from_mime_type(mime_type: str) -> DataFormat:
+    formats = {
+        "application/json": DataFormat.JSON,
+        "application/octet-stream": DataFormat.JSON,  # default in aiohttp
+    }
+    if mime_type not in formats:
+        raise DataParsingError(f"unsupported MIME type '{mime_type}', expected: {str(formats)[1:-1]}")
+    return formats[mime_type]
+
+
+def parse_from_mime_type(data: str, mime_type: str) -> Any:
+    return from_mime_type(mime_type).parse_to_dict(data)
+
+
 class Server:
     # pylint: disable=too-many-instance-attributes
     # This is top-level class containing pretty much everything. Instead of global
@@ -181,7 +195,7 @@ class Server:
         if request.method == "GET":
             update_with: Optional[Dict[str, Any]] = None
         else:
-            update_with = parse(await request.text(), request.content_type)
+            update_with = parse_from_mime_type(await request.text(), request.content_type)
         document_path = request.match_info["path"]
         getheaders = ignore_exceptions_optional(List[str], None, KeyError)(request.headers.getall)
         etags = getheaders("if-match")

manager/knot_resolver_manager/utils/modeling/__init__.py

 from .base_schema import BaseSchema, ConfigSchema
 from .base_value_type import BaseValueType
-from .parsing import parse, parse_json, parse_yaml, try_to_parse
+from .parsing import parse_json, parse_yaml, try_to_parse
 
 __all__ = [
     "BaseValueType",
     "BaseSchema",
     "ConfigSchema",
-    "parse",
     "parse_yaml",
     "parse_json",
     "try_to_parse",

manager/knot_resolver_manager/utils/modeling/parsing.py

@@ -49,53 +49,35 @@ class _RaiseDuplicatesLoader(yaml.SafeLoader):
         return mapping
 
 
-class _Format(Enum):
+class DataFormat(Enum):
     YAML = auto()
     JSON = auto()
 
     def parse_to_dict(self, text: str) -> Any:
-        if self is _Format.YAML:
+        if self is DataFormat.YAML:
             # RaiseDuplicatesLoader extends yaml.SafeLoader, so this should be safe
             # https://python.land/data-processing/python-yaml#PyYAML_safe_load_vs_load
             return renamed(yaml.load(text, Loader=_RaiseDuplicatesLoader))  # type: ignore
-        elif self is _Format.JSON:
+        elif self is DataFormat.JSON:
             return renamed(json.loads(text, object_pairs_hook=_json_raise_duplicates))
         else:
             raise NotImplementedError(f"Parsing of format '{self}' is not implemented")
 
     def dict_dump(self, data: Dict[str, Any]) -> str:
-        if self is _Format.YAML:
+        if self is DataFormat.YAML:
             return yaml.safe_dump(data)  # type: ignore
-        elif self is _Format.JSON:
+        elif self is DataFormat.JSON:
             return json.dumps(data)
         else:
             raise NotImplementedError(f"Exporting to '{self}' format is not implemented")
 
-    @staticmethod
-    def from_mime_type(mime_type: str) -> "_Format":
-        formats = {
-            "application/json": _Format.JSON,
-            "application/yaml": _Format.YAML,
-            "application/octet-stream": _Format.JSON,  # default in aiohttp
-            "text/vnd.yaml": _Format.YAML,
-        }
-        if mime_type not in formats:
-            raise DataParsingError(
-                f"unsupported MIME type '{mime_type}', expected 'application/json' or 'application/yaml'"
-            )
-        return formats[mime_type]
-
-
-def parse(data: str, mime_type: str) -> Any:
-    return _Format.from_mime_type(mime_type).parse_to_dict(data)
-
 
 def parse_yaml(data: str) -> Any:
-    return _Format.YAML.parse_to_dict(data)
+    return DataFormat.YAML.parse_to_dict(data)
 
 
 def parse_json(data: str) -> Any:
-    return _Format.JSON.parse_to_dict(data)
+    return DataFormat.JSON.parse_to_dict(data)
 
 
 def try_to_parse(data: str) -> Any: