Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _apply_validators(
self, v: Any, values: Dict[str, Any], loc: 'LocStr', cls: Optional['ModelOrDc'], validators: 'ValidatorsList'
) -> 'ValidateReturn':
for validator in validators:
try:
v = validator(cls, v, values, self, self.model_config)
except (ValueError, TypeError, AssertionError) as exc:
return v, ErrorWrapper(exc, loc)
return v, None
for name, field in model.__fields__.items():
if type(field.type_) == ForwardRef:
raise ConfigError(
f'field "{field.name}" not yet prepared so type is still a ForwardRef, '
f'you might need to call {cls_.__name__}.update_forward_refs().'
)
value = input_data.get(field.alias, _missing)
using_name = False
if value is _missing and config.allow_population_by_field_name and field.alt_alias:
value = input_data.get(field.name, _missing)
using_name = True
if value is _missing:
if field.required:
errors.append(ErrorWrapper(MissingError(), loc=field.alias))
continue
if field.default is None:
# deepcopy is quite slow on None
value = None
else:
value = deepcopy(field.default)
if not config.validate_all and not field.validate_always:
values[name] = value
continue
else:
fields_set.add(name)
if check_extra:
names_used.add(field.name if using_name else field.alias)
for field in required_params:
if is_scalar_sequence_field(field) and isinstance(
received_params, (QueryParams, Headers)
):
value = received_params.getlist(field.alias) or field.default
else:
value = received_params.get(field.alias)
field_info = get_field_info(field)
assert isinstance(
field_info, params.Param
), "Params must be subclasses of Param"
if value is None:
if field.required:
if PYDANTIC_1:
errors.append(
ErrorWrapper(
MissingError(), loc=(field_info.in_.value, field.alias)
)
)
else: # pragma: nocover
errors.append(
ErrorWrapper( # type: ignore
MissingError(),
loc=(field_info.in_.value, field.alias),
config=BaseConfig,
)
)
else:
values[field.name] = deepcopy(field.default)
continue
v_, errors_ = field.validate(
value, values, loc=(field_info.in_.value, field.alias)
value = received_params.get(field.alias)
field_info = get_field_info(field)
assert isinstance(
field_info, params.Param
), "Params must be subclasses of Param"
if value is None:
if field.required:
if PYDANTIC_1:
errors.append(
ErrorWrapper(
MissingError(), loc=(field_info.in_.value, field.alias)
)
)
else: # pragma: nocover
errors.append(
ErrorWrapper( # type: ignore
MissingError(),
loc=(field_info.in_.value, field.alias),
config=BaseConfig,
)
)
else:
values[field.name] = deepcopy(field.default)
continue
v_, errors_ = field.validate(
value, values, loc=(field_info.in_.value, field.alias)
)
if isinstance(errors_, ErrorWrapper):
errors.append(errors_)
elif isinstance(errors_, list):
errors.extend(errors_)
else:
"""
Validate sequence-like containers: lists, tuples, sets and generators
Note that large if-else blocks are necessary to enable Cython
optimization, which is why we disable the complexity check above.
"""
if not sequence_like(v):
e: errors_.PydanticTypeError
if self.shape == SHAPE_LIST:
e = errors_.ListError()
elif self.shape == SHAPE_SET:
e = errors_.SetError()
elif self.shape == SHAPE_FROZENSET:
e = errors_.FrozenSetError()
else:
e = errors_.SequenceError()
return v, ErrorWrapper(e, loc)
loc = loc if isinstance(loc, tuple) else (loc,)
result = []
errors: List[ErrorList] = []
for i, v_ in enumerate(v):
v_loc = *loc, i
r, ee = self._validate_singleton(v_, values, v_loc, cls)
if ee:
errors.append(ee)
else:
result.append(r)
if errors:
return v, errors
converted: Union[List[Any], Set[Any], FrozenSet[Any], Tuple[Any, ...], Iterator[Any]] = result
"""
values = {}
errors = []
# input_data names, possibly alias
names_used = set()
# field names, never aliases
fields_set = set()
config = model.__config__
check_extra = config.extra is not Extra.ignore
cls_ = cls or model
for validator in model.__pre_root_validators__:
try:
input_data = validator(cls_, input_data)
except (ValueError, TypeError, AssertionError) as exc:
return {}, set(), ValidationError([ErrorWrapper(exc, loc=ROOT_KEY)], cls_)
for name, field in model.__fields__.items():
if type(field.type_) == ForwardRef:
raise ConfigError(
f'field "{field.name}" not yet prepared so type is still a ForwardRef, '
f'you might need to call {cls_.__name__}.update_forward_refs().'
)
value = input_data.get(field.alias, _missing)
using_name = False
if value is _missing and config.allow_population_by_field_name and field.alt_alias:
value = input_data.get(field.name, _missing)
using_name = True
if value is _missing:
if field.required:
content_type: str = None,
encoding: str = 'utf8',
proto: Protocol = None,
allow_pickle: bool = False,
) -> 'Model':
try:
obj = load_str_bytes(
b,
proto=proto,
content_type=content_type,
encoding=encoding,
allow_pickle=allow_pickle,
json_loads=cls.__config__.json_loads,
)
except (ValueError, TypeError, UnicodeDecodeError) as e:
raise ValidationError([ErrorWrapper(e, loc=ROOT_KEY)], cls)
return cls.parse_obj(obj)
def _validate_mapping(
self, v: Any, values: Dict[str, Any], loc: 'LocStr', cls: Optional['ModelOrDc']
) -> 'ValidateReturn':
try:
v_iter = dict_validator(v)
except TypeError as exc:
return v, ErrorWrapper(exc, loc)
loc = loc if isinstance(loc, tuple) else (loc,)
result, errors = {}, []
for k, v_ in v_iter.items():
v_loc = *loc, '__key__'
key_result, key_errors = self.key_field.validate(k, values, loc=v_loc, cls=cls) # type: ignore
if key_errors:
errors.append(key_errors)
continue
v_loc = *loc, k
value_result, value_errors = self._validate_singleton(v_, values, v_loc, cls)
if value_errors:
errors.append(value_errors)
continue
resource_type is not None
and resource_type != __pydantic_self__.__fields__["resource_type"].default
):
expected_resource_type = __pydantic_self__.__fields__[
"resource_type"
].default
error = (
f"``{__pydantic_self__.__class__.__module__}."
f"{__pydantic_self__.__class__.__name__}`` "
f"expects resource type ``{expected_resource_type}``, "
f"but got ``{resource_type}``. "
"Make sure resource type name is correct and right "
"ModelClass has been chosen."
)
errors.append(
ErrorWrapper(WrongResourceType(error=error), loc="resource_type")
)
if errors:
raise ValidationError(errors, __pydantic_self__.__class__)
BaseModel.__init__(__pydantic_self__, **data)
else:
values[name] = v_
if check_extra:
if isinstance(input_data, GetterDict):
extra = input_data.extra_keys() - names_used
else:
extra = input_data.keys() - names_used
if extra:
fields_set |= extra
if config.extra is Extra.allow:
for f in extra:
values[f] = input_data[f]
else:
for f in sorted(extra):
errors.append(ErrorWrapper(ExtraError(), loc=f))
for skip_on_failure, validator in model.__post_root_validators__:
if skip_on_failure and errors:
continue
try:
values = validator(cls_, values)
except (ValueError, TypeError, AssertionError) as exc:
errors.append(ErrorWrapper(exc, loc=ROOT_KEY))
break
if errors:
return values, fields_set, ValidationError(errors, cls_)
else:
return values, fields_set, None