(
body_fields: list[ModelField],
received_body: FormData,
)
| 914 | |
| 915 | |
| 916 | async def _extract_form_body( |
| 917 | body_fields: list[ModelField], |
| 918 | received_body: FormData, |
| 919 | ) -> dict[str, Any]: |
| 920 | values = {} |
| 921 | |
| 922 | for field in body_fields: |
| 923 | value = _get_multidict_value(field, received_body) |
| 924 | field_info = field.field_info |
| 925 | if ( |
| 926 | isinstance(field_info, params.File) |
| 927 | and is_bytes_or_nonable_bytes_annotation(field.field_info.annotation) |
| 928 | and isinstance(value, UploadFile) |
| 929 | ): |
| 930 | value = await value.read() |
| 931 | elif ( |
| 932 | is_bytes_sequence_annotation(field.field_info.annotation) |
| 933 | and isinstance(field_info, params.File) |
| 934 | and value_is_sequence(value) |
| 935 | ): |
| 936 | # For types |
| 937 | assert isinstance(value, sequence_types) |
| 938 | results: list[bytes | str] = [] |
| 939 | for sub_value in value: |
| 940 | results.append(await sub_value.read()) |
| 941 | value = serialize_sequence_value(field=field, value=results) |
| 942 | if value is not None: |
| 943 | values[get_validation_alias(field)] = value |
| 944 | field_aliases = {get_validation_alias(field) for field in body_fields} |
| 945 | for key in received_body.keys(): |
| 946 | if key not in field_aliases: |
| 947 | param_values = received_body.getlist(key) |
| 948 | if len(param_values) == 1: |
| 949 | values[key] = param_values[0] |
| 950 | else: |
| 951 | values[key] = param_values |
| 952 | return values |
| 953 | |
| 954 | |
| 955 | async def request_body_to_args( |
no test coverage detected
searching dependent graphs…