(
fields: Sequence[ModelField],
received_params: Mapping[str, Any] | QueryParams | Headers,
)
| 782 | |
| 783 | |
| 784 | def request_params_to_args( |
| 785 | fields: Sequence[ModelField], |
| 786 | received_params: Mapping[str, Any] | QueryParams | Headers, |
| 787 | ) -> tuple[dict[str, Any], list[Any]]: |
| 788 | values: dict[str, Any] = {} |
| 789 | errors: list[dict[str, Any]] = [] |
| 790 | |
| 791 | if not fields: |
| 792 | return values, errors |
| 793 | |
| 794 | first_field = fields[0] |
| 795 | fields_to_extract = fields |
| 796 | single_not_embedded_field = False |
| 797 | default_convert_underscores = True |
| 798 | if len(fields) == 1 and lenient_issubclass( |
| 799 | first_field.field_info.annotation, BaseModel |
| 800 | ): |
| 801 | fields_to_extract = get_cached_model_fields(first_field.field_info.annotation) |
| 802 | single_not_embedded_field = True |
| 803 | # If headers are in a Pydantic model, the way to disable convert_underscores |
| 804 | # would be with Header(convert_underscores=False) at the Pydantic model level |
| 805 | default_convert_underscores = getattr( |
| 806 | first_field.field_info, "convert_underscores", True |
| 807 | ) |
| 808 | |
| 809 | params_to_process: dict[str, Any] = {} |
| 810 | |
| 811 | processed_keys = set() |
| 812 | |
| 813 | for field in fields_to_extract: |
| 814 | alias = None |
| 815 | if isinstance(received_params, Headers): |
| 816 | # Handle fields extracted from a Pydantic Model for a header, each field |
| 817 | # doesn't have a FieldInfo of type Header with the default convert_underscores=True |
| 818 | convert_underscores = getattr( |
| 819 | field.field_info, "convert_underscores", default_convert_underscores |
| 820 | ) |
| 821 | if convert_underscores: |
| 822 | alias = get_validation_alias(field) |
| 823 | if alias == field.name: |
| 824 | alias = alias.replace("_", "-") |
| 825 | value = _get_multidict_value(field, received_params, alias=alias) |
| 826 | if value is not None: |
| 827 | params_to_process[get_validation_alias(field)] = value |
| 828 | processed_keys.add(alias or get_validation_alias(field)) |
| 829 | # For headers with convert_underscores=True, mark both the converted |
| 830 | # header name and the original field alias as processed to avoid |
| 831 | # accepting the original alias as an extra header. |
| 832 | processed_keys.add(get_validation_alias(field)) |
| 833 | |
| 834 | for key in received_params.keys(): |
| 835 | if key not in processed_keys: |
| 836 | if isinstance(received_params, (ImmutableMultiDict, Headers)): |
| 837 | value = received_params.getlist(key) |
| 838 | if isinstance(value, list) and (len(value) == 1): |
| 839 | params_to_process[key] = value[0] |
| 840 | else: |
| 841 | params_to_process[key] = value |
no test coverage detected
searching dependent graphs…