|
|
|
@ -181,13 +181,13 @@ def get_flat_dependant( |
|
|
|
def _get_flat_fields_from_params(fields: list[ModelField]) -> list[ModelField]: |
|
|
|
if not fields: |
|
|
|
return fields |
|
|
|
first_field = fields[0] |
|
|
|
if len(fields) == 1 and lenient_issubclass( |
|
|
|
first_field.field_info.annotation, BaseModel |
|
|
|
): |
|
|
|
fields_to_extract = get_cached_model_fields(first_field.field_info.annotation) |
|
|
|
return fields_to_extract |
|
|
|
return fields |
|
|
|
fields_to_extract = [] |
|
|
|
for f in fields: |
|
|
|
if lenient_issubclass(f.field_info.annotation, BaseModel): |
|
|
|
fields_to_extract.extend(get_cached_model_fields(f.field_info.annotation)) |
|
|
|
else: |
|
|
|
fields_to_extract.append(f) |
|
|
|
return fields_to_extract |
|
|
|
|
|
|
|
|
|
|
|
def get_flat_params(dependant: Dependant) -> list[ModelField]: |
|
|
|
@ -762,32 +762,25 @@ def request_params_to_args( |
|
|
|
if not fields: |
|
|
|
return values, errors |
|
|
|
|
|
|
|
first_field = fields[0] |
|
|
|
fields_to_extract = fields |
|
|
|
single_not_embedded_field = False |
|
|
|
default_convert_underscores = True |
|
|
|
if len(fields) == 1 and lenient_issubclass( |
|
|
|
first_field.field_info.annotation, BaseModel |
|
|
|
): |
|
|
|
fields_to_extract = get_cached_model_fields(first_field.field_info.annotation) |
|
|
|
single_not_embedded_field = True |
|
|
|
# If headers are in a Pydantic model, the way to disable convert_underscores |
|
|
|
# would be with Header(convert_underscores=False) at the Pydantic model level |
|
|
|
default_convert_underscores = getattr( |
|
|
|
first_field.field_info, "convert_underscores", True |
|
|
|
) |
|
|
|
|
|
|
|
params_to_process: dict[str, Any] = {} |
|
|
|
|
|
|
|
fields_to_extract = [ |
|
|
|
(field, cached_field) |
|
|
|
for field in fields |
|
|
|
if lenient_issubclass(field.field_info.annotation, BaseModel) |
|
|
|
for cached_field in get_cached_model_fields(field.field_info.annotation) |
|
|
|
] |
|
|
|
|
|
|
|
processed_keys = set() |
|
|
|
|
|
|
|
for field in fields_to_extract: |
|
|
|
for parent_field, field in fields_to_extract: |
|
|
|
alias = None |
|
|
|
if isinstance(received_params, Headers): |
|
|
|
# Handle fields extracted from a Pydantic Model for a header, each field |
|
|
|
# doesn't have a FieldInfo of type Header with the default convert_underscores=True |
|
|
|
convert_underscores = getattr( |
|
|
|
field.field_info, "convert_underscores", default_convert_underscores |
|
|
|
parent_field.field_info, "convert_underscores", default_convert_underscores |
|
|
|
) |
|
|
|
if convert_underscores: |
|
|
|
alias = get_validation_alias(field) |
|
|
|
@ -809,27 +802,21 @@ def request_params_to_args( |
|
|
|
else: |
|
|
|
params_to_process[key] = received_params.get(key) |
|
|
|
|
|
|
|
if single_not_embedded_field: |
|
|
|
field_info = first_field.field_info |
|
|
|
assert isinstance(field_info, params.Param), ( |
|
|
|
"Params must be subclasses of Param" |
|
|
|
) |
|
|
|
loc: tuple[str, ...] = (field_info.in_.value,) |
|
|
|
v_, errors_ = _validate_value_with_model_field( |
|
|
|
field=first_field, value=params_to_process, values=values, loc=loc |
|
|
|
) |
|
|
|
return {first_field.name: v_}, errors_ |
|
|
|
|
|
|
|
for field in fields: |
|
|
|
value = _get_multidict_value(field, received_params) |
|
|
|
field_info = field.field_info |
|
|
|
assert isinstance(field_info, params.Param), ( |
|
|
|
"Params must be subclasses of Param" |
|
|
|
) |
|
|
|
loc = (field_info.in_.value, get_validation_alias(field)) |
|
|
|
v_, errors_ = _validate_value_with_model_field( |
|
|
|
field=field, value=value, values=values, loc=loc |
|
|
|
) |
|
|
|
in_ = getattr(field.field_info, "in_", params.ParamTypes.query) |
|
|
|
|
|
|
|
if lenient_issubclass(field.field_info.annotation, BaseModel): |
|
|
|
loc: tuple[str, ...] = (in_.value,) |
|
|
|
v_, errors_ = _validate_value_with_model_field( |
|
|
|
field=field, value=params_to_process, values=values, loc=loc |
|
|
|
) |
|
|
|
else: |
|
|
|
value = _get_multidict_value(field, received_params) |
|
|
|
loc = (in_.value, field.alias) |
|
|
|
v_, errors_ = _validate_value_with_model_field( |
|
|
|
field=field, value=value, values=values, loc=loc |
|
|
|
) |
|
|
|
|
|
|
|
if errors_: |
|
|
|
errors.extend(errors_) |
|
|
|
else: |
|
|
|
|