| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161 |
- import json
- from typing import TYPE_CHECKING, Any, Dict, Generator, List, Optional, Sequence, Tuple, Type, Union
- from pydantic.v1.json import pydantic_encoder
- from pydantic.v1.utils import Representation
- if TYPE_CHECKING:
- from typing_extensions import TypedDict
- from pydantic.v1.config import BaseConfig
- from pydantic.v1.types import ModelOrDc
- from pydantic.v1.typing import ReprArgs
- Loc = Tuple[Union[int, str], ...]
- class _ErrorDictRequired(TypedDict):
- loc: Loc
- msg: str
- type: str
- class ErrorDict(_ErrorDictRequired, total=False):
- ctx: Dict[str, Any]
- __all__ = 'ErrorWrapper', 'ValidationError'
- class ErrorWrapper(Representation):
- __slots__ = 'exc', '_loc'
- def __init__(self, exc: Exception, loc: Union[str, 'Loc']) -> None:
- self.exc = exc
- self._loc = loc
- def loc_tuple(self) -> 'Loc':
- if isinstance(self._loc, tuple):
- return self._loc
- else:
- return (self._loc,)
- def __repr_args__(self) -> 'ReprArgs':
- return [('exc', self.exc), ('loc', self.loc_tuple())]
- # ErrorList is something like Union[List[Union[List[ErrorWrapper], ErrorWrapper]], ErrorWrapper]
- # but recursive, therefore just use:
- ErrorList = Union[Sequence[Any], ErrorWrapper]
- class ValidationError(Representation, ValueError):
- __slots__ = 'raw_errors', 'model', '_error_cache'
- def __init__(self, errors: Sequence[ErrorList], model: 'ModelOrDc') -> None:
- self.raw_errors = errors
- self.model = model
- self._error_cache: Optional[List['ErrorDict']] = None
- def errors(self) -> List['ErrorDict']:
- if self._error_cache is None:
- try:
- config = self.model.__config__ # type: ignore
- except AttributeError:
- config = self.model.__pydantic_model__.__config__ # type: ignore
- self._error_cache = list(flatten_errors(self.raw_errors, config))
- return self._error_cache
- def json(self, *, indent: Union[None, int, str] = 2) -> str:
- return json.dumps(self.errors(), indent=indent, default=pydantic_encoder)
- def __str__(self) -> str:
- errors = self.errors()
- no_errors = len(errors)
- return (
- f'{no_errors} validation error{"" if no_errors == 1 else "s"} for {self.model.__name__}\n'
- f'{display_errors(errors)}'
- )
- def __repr_args__(self) -> 'ReprArgs':
- return [('model', self.model.__name__), ('errors', self.errors())]
- def display_errors(errors: List['ErrorDict']) -> str:
- return '\n'.join(f'{_display_error_loc(e)}\n {e["msg"]} ({_display_error_type_and_ctx(e)})' for e in errors)
- def _display_error_loc(error: 'ErrorDict') -> str:
- return ' -> '.join(str(e) for e in error['loc'])
- def _display_error_type_and_ctx(error: 'ErrorDict') -> str:
- t = 'type=' + error['type']
- ctx = error.get('ctx')
- if ctx:
- return t + ''.join(f'; {k}={v}' for k, v in ctx.items())
- else:
- return t
- def flatten_errors(
- errors: Sequence[Any], config: Type['BaseConfig'], loc: Optional['Loc'] = None
- ) -> Generator['ErrorDict', None, None]:
- for error in errors:
- if isinstance(error, ErrorWrapper):
- if loc:
- error_loc = loc + error.loc_tuple()
- else:
- error_loc = error.loc_tuple()
- if isinstance(error.exc, ValidationError):
- yield from flatten_errors(error.exc.raw_errors, config, error_loc)
- else:
- yield error_dict(error.exc, config, error_loc)
- elif isinstance(error, list):
- yield from flatten_errors(error, config, loc=loc)
- else:
- raise RuntimeError(f'Unknown error object: {error}')
- def error_dict(exc: Exception, config: Type['BaseConfig'], loc: 'Loc') -> 'ErrorDict':
- type_ = get_exc_type(exc.__class__)
- msg_template = config.error_msg_templates.get(type_) or getattr(exc, 'msg_template', None)
- ctx = exc.__dict__
- if msg_template:
- msg = msg_template.format(**ctx)
- else:
- msg = str(exc)
- d: 'ErrorDict' = {'loc': loc, 'msg': msg, 'type': type_}
- if ctx:
- d['ctx'] = ctx
- return d
- _EXC_TYPE_CACHE: Dict[Type[Exception], str] = {}
- def get_exc_type(cls: Type[Exception]) -> str:
- # slightly more efficient than using lru_cache since we don't need to worry about the cache filling up
- try:
- return _EXC_TYPE_CACHE[cls]
- except KeyError:
- r = _get_exc_type(cls)
- _EXC_TYPE_CACHE[cls] = r
- return r
- def _get_exc_type(cls: Type[Exception]) -> str:
- if issubclass(cls, AssertionError):
- return 'assertion_error'
- base_name = 'type_error' if issubclass(cls, TypeError) else 'value_error'
- if cls in (TypeError, ValueError):
- # just TypeError or ValueError, no extra code
- return base_name
- # if it's not a TypeError or ValueError, we just take the lowercase of the exception name
- # no chaining or snake case logic, use "code" for more complex error types.
- code = getattr(cls, 'code', None) or cls.__name__.replace('Error', '').lower()
- return base_name + '.' + code
|