You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
318 lines
11 KiB
318 lines
11 KiB
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from contextlib import contextmanager
|
|
from textwrap import indent, wrap
|
|
from typing import Any, Dict, Iterator, List, Optional, Sequence, Union, cast
|
|
|
|
from .fastjsonschema_exceptions import JsonSchemaValueException
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
_MESSAGE_REPLACEMENTS = {
|
|
"must be named by propertyName definition": "keys must be named by",
|
|
"one of contains definition": "at least one item that matches",
|
|
" same as const definition:": "",
|
|
"only specified items": "only items matching the definition",
|
|
}
|
|
|
|
_SKIP_DETAILS = (
|
|
"must not be empty",
|
|
"is always invalid",
|
|
"must not be there",
|
|
)
|
|
|
|
_NEED_DETAILS = {"anyOf", "oneOf", "anyOf", "contains", "propertyNames", "not", "items"}
|
|
|
|
_CAMEL_CASE_SPLITTER = re.compile(r"\W+|([A-Z][^A-Z\W]*)")
|
|
_IDENTIFIER = re.compile(r"^[\w_]+$", re.I)
|
|
|
|
_TOML_JARGON = {
|
|
"object": "table",
|
|
"property": "key",
|
|
"properties": "keys",
|
|
"property names": "keys",
|
|
}
|
|
|
|
|
|
class ValidationError(JsonSchemaValueException):
|
|
"""Report violations of a given JSON schema.
|
|
|
|
This class extends :exc:`~fastjsonschema.JsonSchemaValueException`
|
|
by adding the following properties:
|
|
|
|
- ``summary``: an improved version of the ``JsonSchemaValueException`` error message
|
|
with only the necessary information)
|
|
|
|
- ``details``: more contextual information about the error like the failing schema
|
|
itself and the value that violates the schema.
|
|
|
|
Depending on the level of the verbosity of the ``logging`` configuration
|
|
the exception message will be only ``summary`` (default) or a combination of
|
|
``summary`` and ``details`` (when the logging level is set to :obj:`logging.DEBUG`).
|
|
"""
|
|
|
|
summary = ""
|
|
details = ""
|
|
_original_message = ""
|
|
|
|
@classmethod
|
|
def _from_jsonschema(cls, ex: JsonSchemaValueException):
|
|
formatter = _ErrorFormatting(ex)
|
|
obj = cls(str(formatter), ex.value, formatter.name, ex.definition, ex.rule)
|
|
debug_code = os.getenv("JSONSCHEMA_DEBUG_CODE_GENERATION", "false").lower()
|
|
if debug_code != "false": # pragma: no cover
|
|
obj.__cause__, obj.__traceback__ = ex.__cause__, ex.__traceback__
|
|
obj._original_message = ex.message
|
|
obj.summary = formatter.summary
|
|
obj.details = formatter.details
|
|
return obj
|
|
|
|
|
|
@contextmanager
|
|
def detailed_errors():
|
|
try:
|
|
yield
|
|
except JsonSchemaValueException as ex:
|
|
raise ValidationError._from_jsonschema(ex) from None
|
|
|
|
|
|
class _ErrorFormatting:
|
|
def __init__(self, ex: JsonSchemaValueException):
|
|
self.ex = ex
|
|
self.name = f"`{self._simplify_name(ex.name)}`"
|
|
self._original_message = self.ex.message.replace(ex.name, self.name)
|
|
self._summary = ""
|
|
self._details = ""
|
|
|
|
def __str__(self) -> str:
|
|
if _logger.getEffectiveLevel() <= logging.DEBUG and self.details:
|
|
return f"{self.summary}\n\n{self.details}"
|
|
|
|
return self.summary
|
|
|
|
@property
|
|
def summary(self) -> str:
|
|
if not self._summary:
|
|
self._summary = self._expand_summary()
|
|
|
|
return self._summary
|
|
|
|
@property
|
|
def details(self) -> str:
|
|
if not self._details:
|
|
self._details = self._expand_details()
|
|
|
|
return self._details
|
|
|
|
def _simplify_name(self, name):
|
|
x = len("data.")
|
|
return name[x:] if name.startswith("data.") else name
|
|
|
|
def _expand_summary(self):
|
|
msg = self._original_message
|
|
|
|
for bad, repl in _MESSAGE_REPLACEMENTS.items():
|
|
msg = msg.replace(bad, repl)
|
|
|
|
if any(substring in msg for substring in _SKIP_DETAILS):
|
|
return msg
|
|
|
|
schema = self.ex.rule_definition
|
|
if self.ex.rule in _NEED_DETAILS and schema:
|
|
summary = _SummaryWriter(_TOML_JARGON)
|
|
return f"{msg}:\n\n{indent(summary(schema), ' ')}"
|
|
|
|
return msg
|
|
|
|
def _expand_details(self) -> str:
|
|
optional = []
|
|
desc_lines = self.ex.definition.pop("$$description", [])
|
|
desc = self.ex.definition.pop("description", None) or " ".join(desc_lines)
|
|
if desc:
|
|
description = "\n".join(
|
|
wrap(
|
|
desc,
|
|
width=80,
|
|
initial_indent=" ",
|
|
subsequent_indent=" ",
|
|
break_long_words=False,
|
|
)
|
|
)
|
|
optional.append(f"DESCRIPTION:\n{description}")
|
|
schema = json.dumps(self.ex.definition, indent=4)
|
|
value = json.dumps(self.ex.value, indent=4)
|
|
defaults = [
|
|
f"GIVEN VALUE:\n{indent(value, ' ')}",
|
|
f"OFFENDING RULE: {self.ex.rule!r}",
|
|
f"DEFINITION:\n{indent(schema, ' ')}",
|
|
]
|
|
return "\n\n".join(optional + defaults)
|
|
|
|
|
|
class _SummaryWriter:
|
|
_IGNORE = {"description", "default", "title", "examples"}
|
|
|
|
def __init__(self, jargon: Optional[Dict[str, str]] = None):
|
|
self.jargon: Dict[str, str] = jargon or {}
|
|
# Clarify confusing terms
|
|
self._terms = {
|
|
"anyOf": "at least one of the following",
|
|
"oneOf": "exactly one of the following",
|
|
"allOf": "all of the following",
|
|
"not": "(*NOT* the following)",
|
|
"prefixItems": f"{self._jargon('items')} (in order)",
|
|
"items": "items",
|
|
"contains": "contains at least one of",
|
|
"propertyNames": (
|
|
f"non-predefined acceptable {self._jargon('property names')}"
|
|
),
|
|
"patternProperties": f"{self._jargon('properties')} named via pattern",
|
|
"const": "predefined value",
|
|
"enum": "one of",
|
|
}
|
|
# Attributes that indicate that the definition is easy and can be done
|
|
# inline (e.g. string and number)
|
|
self._guess_inline_defs = [
|
|
"enum",
|
|
"const",
|
|
"maxLength",
|
|
"minLength",
|
|
"pattern",
|
|
"format",
|
|
"minimum",
|
|
"maximum",
|
|
"exclusiveMinimum",
|
|
"exclusiveMaximum",
|
|
"multipleOf",
|
|
]
|
|
|
|
def _jargon(self, term: Union[str, List[str]]) -> Union[str, List[str]]:
|
|
if isinstance(term, list):
|
|
return [self.jargon.get(t, t) for t in term]
|
|
return self.jargon.get(term, term)
|
|
|
|
def __call__(
|
|
self,
|
|
schema: Union[dict, List[dict]],
|
|
prefix: str = "",
|
|
*,
|
|
_path: Sequence[str] = (),
|
|
) -> str:
|
|
if isinstance(schema, list):
|
|
return self._handle_list(schema, prefix, _path)
|
|
|
|
filtered = self._filter_unecessary(schema, _path)
|
|
simple = self._handle_simple_dict(filtered, _path)
|
|
if simple:
|
|
return f"{prefix}{simple}"
|
|
|
|
child_prefix = self._child_prefix(prefix, " ")
|
|
item_prefix = self._child_prefix(prefix, "- ")
|
|
indent = len(prefix) * " "
|
|
with io.StringIO() as buffer:
|
|
for i, (key, value) in enumerate(filtered.items()):
|
|
child_path = [*_path, key]
|
|
line_prefix = prefix if i == 0 else indent
|
|
buffer.write(f"{line_prefix}{self._label(child_path)}:")
|
|
# ^ just the first item should receive the complete prefix
|
|
if isinstance(value, dict):
|
|
filtered = self._filter_unecessary(value, child_path)
|
|
simple = self._handle_simple_dict(filtered, child_path)
|
|
buffer.write(
|
|
f" {simple}"
|
|
if simple
|
|
else f"\n{self(value, child_prefix, _path=child_path)}"
|
|
)
|
|
elif isinstance(value, list) and (
|
|
key != "type" or self._is_property(child_path)
|
|
):
|
|
children = self._handle_list(value, item_prefix, child_path)
|
|
sep = " " if children.startswith("[") else "\n"
|
|
buffer.write(f"{sep}{children}")
|
|
else:
|
|
buffer.write(f" {self._value(value, child_path)}\n")
|
|
return buffer.getvalue()
|
|
|
|
def _is_unecessary(self, path: Sequence[str]) -> bool:
|
|
if self._is_property(path) or not path: # empty path => instruction @ root
|
|
return False
|
|
key = path[-1]
|
|
return any(key.startswith(k) for k in "$_") or key in self._IGNORE
|
|
|
|
def _filter_unecessary(self, schema: dict, path: Sequence[str]):
|
|
return {
|
|
key: value
|
|
for key, value in schema.items()
|
|
if not self._is_unecessary([*path, key])
|
|
}
|
|
|
|
def _handle_simple_dict(self, value: dict, path: Sequence[str]) -> Optional[str]:
|
|
inline = any(p in value for p in self._guess_inline_defs)
|
|
simple = not any(isinstance(v, (list, dict)) for v in value.values())
|
|
if inline or simple:
|
|
return f"{{{', '.join(self._inline_attrs(value, path))}}}\n"
|
|
return None
|
|
|
|
def _handle_list(
|
|
self, schemas: list, prefix: str = "", path: Sequence[str] = ()
|
|
) -> str:
|
|
if self._is_unecessary(path):
|
|
return ""
|
|
|
|
repr_ = repr(schemas)
|
|
if all(not isinstance(e, (dict, list)) for e in schemas) and len(repr_) < 60:
|
|
return f"{repr_}\n"
|
|
|
|
item_prefix = self._child_prefix(prefix, "- ")
|
|
return "".join(
|
|
self(v, item_prefix, _path=[*path, f"[{i}]"]) for i, v in enumerate(schemas)
|
|
)
|
|
|
|
def _is_property(self, path: Sequence[str]):
|
|
"""Check if the given path can correspond to an arbitrarily named property"""
|
|
counter = 0
|
|
for key in path[-2::-1]:
|
|
if key not in {"properties", "patternProperties"}:
|
|
break
|
|
counter += 1
|
|
|
|
# If the counter if even, the path correspond to a JSON Schema keyword
|
|
# otherwise it can be any arbitrary string naming a property
|
|
return counter % 2 == 1
|
|
|
|
def _label(self, path: Sequence[str]) -> str:
|
|
*parents, key = path
|
|
if not self._is_property(path):
|
|
norm_key = _separate_terms(key)
|
|
return self._terms.get(key) or " ".join(self._jargon(norm_key))
|
|
|
|
if parents[-1] == "patternProperties":
|
|
return f"(regex {key!r})"
|
|
return repr(key) # property name
|
|
|
|
def _value(self, value: Any, path: Sequence[str]) -> str:
|
|
if path[-1] == "type" and not self._is_property(path):
|
|
type_ = self._jargon(value)
|
|
return (
|
|
f"[{', '.join(type_)}]" if isinstance(value, list) else cast(str, type_)
|
|
)
|
|
return repr(value)
|
|
|
|
def _inline_attrs(self, schema: dict, path: Sequence[str]) -> Iterator[str]:
|
|
for key, value in schema.items():
|
|
child_path = [*path, key]
|
|
yield f"{self._label(child_path)}: {self._value(value, child_path)}"
|
|
|
|
def _child_prefix(self, parent_prefix: str, child_prefix: str) -> str:
|
|
return len(parent_prefix) * " " + child_prefix
|
|
|
|
|
|
def _separate_terms(word: str) -> List[str]:
|
|
"""
|
|
>>> _separate_terms("FooBar-foo")
|
|
['foo', 'bar', 'foo']
|
|
"""
|
|
return [w.lower() for w in _CAMEL_CASE_SPLITTER.split(word) if w]
|
|
|