fastapi_13537

{"status": "ok", "stdout": "./fastapi/dependencies/utils.py:869:async def _extract_form_body(\n./fastapi/dependencies/utils.py:933:        body_to_process = await _extract_form_body(fields_to_extract, received_body)\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/dependencies/utils.py", "content": "async def _extract_form_body(\n    body_fields: List[ModelField],\n    received_body: FormData,\n) -> Dict[str, Any]:\n    values = {}\n\n    for field in body_fields:\n        value = _get_multidict_value(field, received_body)\n        field_info = field.field_info\n        if (\n            isinstance(field_info, (params.File, temp_pydantic_v1_params.File))\n            and is_bytes_field(field)\n            and isinstance(value, UploadFile)\n        ):\n            value = await value.read()\n        elif (\n            is_bytes_sequence_field(field)\n            and isinstance(field_info, (params.File, temp_pydantic_v1_params.File))\n            and value_is_sequence(value)\n        ):\n            # For types\n            assert isinstance(value, sequence_types)  # type: ignore[arg-type]\n            results: List[Union[bytes, str]] = []\n\n            async def process_fn(\n                fn: Callable[[], Coroutine[Any, Any, Any]],\n            ) -> None:\n                result = await fn()\n                results.append(result)  # noqa: B023\n\n            async with anyio.create_task_group() as tg:\n                for sub_value in value:\n                    tg.start_soon(process_fn, sub_value.read)\n            value = serialize_sequence_value(field=field, value=results)\n        if value is not None:\n            values[field.alias] = value\n    for key, value in received_body.items():\n        if key not in values:\n            values[key] = value\n    return values\n\n\nasync def request_body_to_args(\n    body_fields: List[ModelField],\n    received_body: Optional[Union[Dict[str, Any], FormData]],\n    embed_body_fields: bool,\n) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:\n    values: Dict[str, Any] = {}\n    errors: List[Dict[str, Any]] = []\n    assert body_fields, \"request_body_to_args() should be called with fields\"\n    single_not_embedded_field = len(body_fields) 

{"status": "ok", "filepath": "fastapi/dependencies/utils.py", "content": "        assert isinstance(field_info, (params.Param, temp_pydantic_v1_params.Param)), (\n            \"Params must be subclasses of Param\"\n        )\n        loc: Tuple[str, ...] = (field_info.in_.value,)\n        v_, errors_ = _validate_value_with_model_field(\n            field=first_field, value=params_to_process, values=values, loc=loc\n        )\n        return {first_field.name: v_}, errors_\n\n    for field in fields:\n        value = _get_multidict_value(field, received_params)\n        field_info = field.field_info\n        assert isinstance(field_info, (params.Param, temp_pydantic_v1_params.Param)), (\n            \"Params must be subclasses of Param\"\n        )\n        loc = (field_info.in_.value, field.alias)\n        v_, errors_ = _validate_value_with_model_field(\n            field=field, value=value, values=values, loc=loc\n        )\n        if errors_:\n            errors.extend(errors_)\n        else:\n            values[field.name] = v_\n    return values, errors\n\n\ndef is_union_of_base_models(field_type: Any) -> bool:\n    \"\"\"Check if field type is a Union where all members are BaseModel subclasses.\"\"\"\n    from fastapi.types import UnionType\n\n    origin = get_origin(field_type)\n\n    # Check if it's a Union type (covers both typing.Union and types.UnionType in Python 3.10+)\n    if origin is not Union and origin is not UnionType:\n        return False\n\n    union_args = get_args(field_type)\n\n    for arg in union_args:\n        if not _is_model_class(arg):\n            return False\n\n    return True\n\n\ndef _should_embed_body_fields(fields: List[ModelField]) -> bool:\n    if not fields:\n        return False\n    # More than one dependency could have the same field, it would show up as multiple\n    # fields but it's the same one, so count them by name\n    body_param_names_set = {field.name for field in fields}\n    # A top level field has to be a singl

{"status": "ok", "stdout": "fastapi/dependencies/utils.py:723:def _get_multidict_value(\nfastapi/dependencies/utils.py:788:        value = _get_multidict_value(field, received_params, alias=alias)\nfastapi/dependencies/utils.py:810:        value = _get_multidict_value(field, received_params)\nfastapi/dependencies/utils.py:876:        value = _get_multidict_value(field, received_body)\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/dependencies/utils.py", "content": "def _get_multidict_value(\n    field: ModelField, values: Mapping[str, Any], alias: Union[str, None] = None\n) -> Any:\n    alias = alias or field.alias\n    if is_sequence_field(field) and isinstance(values, (ImmutableMultiDict, Headers)):\n        value = values.getlist(alias)\n    else:\n        value = values.get(alias, None)\n    if (\n        value is None\n        or (\n            isinstance(field.field_info, (params.Form, temp_pydantic_v1_params.Form))\n            and isinstance(value, str)  # For type checks\n            and value == \"\"\n        )\n        or (is_sequence_field(field) and len(value) == 0)\n    ):\n        if field.required:", "start_line": 723, "end_line": 740, "total_lines": 1024, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/dependencies/utils.py", "content": "            return\n        else:\n            return deepcopy(field.default)\n    return value\n\n\ndef request_params_to_args(\n    fields: Sequence[ModelField],\n    received_params: Union[Mapping[str, Any], QueryParams, Headers],\n) -> Tuple[Dict[str, Any], List[Any]]:\n    values: Dict[str, Any] = {}\n    errors: List[Dict[str, Any]] = []\n\n    if not fields:\n        return values, errors", "start_line": 741, "end_line": 755, "total_lines": 1024, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/dependencies/utils.py", "content": "\n    first_field = fields[0]\n    fields_to_extract = fields\n    single_not_embedded_field = False\n    default_convert_underscores = True\n    if len(fields) == 1 and lenient_issubclass(first_field.type_, BaseModel):\n        fields_to_extract = get_cached_model_fields(first_field.type_)\n        single_not_embedded_field = True\n        # If headers are in a Pydantic model, the way to disable convert_underscores\n        # would be with Header(convert_underscores=False) at the Pydantic model level\n        default_convert_underscores = getattr(\n            first_field.field_info, \"convert_underscores\", True\n        )\n\n    params_to_process: Dict[str, Any] = {}\n\n    processed_keys = set()\n\n    for field in fields_to_extract:\n        alias = None\n        if isinstance(received_params, Headers):\n            # Handle fields extracted from a Pydantic Model for a header, each field\n            # doesn't have a FieldInfo of type Header with the default convert_underscores=True\n            convert_underscores = getattr(\n                field.field_info, \"convert_underscores\", default_convert_underscores\n            )\n            if convert_underscores:\n                alias = (\n                    field.alias\n                    if field.alias != field.name\n                    else field.name.replace(\"_\", \"-\")\n                )\n        value = _get_multidict_value(field, received_params, alias=alias)\n        if value is not None:\n            params_to_process[field.name] = value\n        processed_keys.add(alias or field.alias)\n        processed_keys.add(field.name)\n\n    for key, value in received_params.items():\n        if key not in processed_keys:\n            params_to_process[key] = value\n\n    if single_not_embedded_field:\n        field_info = first_field.field_info\n        assert isinstance(field_info, (params.Param, temp_pydantic_v1_params.Param)), ("

{"status": "error", "error_type": "FileWriteError", "error_message": "Path traversal detected: '/tmp/repro.py' escapes workspace root."}

{"status": "ok", "filepath": "repro.py", "size": 765}

{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"/private/workspace/repro.py\", line 27, in <module>\n    test_empty_string_int_form_value()\n    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^\n  File \"/private/workspace/repro.py\", line 23, in test_empty_string_int_form_value\n    assert response.status_code == 200\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "details": {"stdout": "Response: {'detail': [{'type': 'int_parsing', 'loc': ['body', 'age'], 'msg': 'Input should be a valid integer, unable to parse string as an integer', 'input': ''}]}\n", "stderr": "Traceback (most recent call last):\n  File \"/private/workspace/repro.py\", line 27, in <module>\n    test_empty_string_int_form_value()\n    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^\n  File \"/private/workspace/repro.py\", line 23, in test_empty_string_int_form_value\n    assert response.status_code == 200\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "exit_code": 1}}

{"status": "ok", "filepath": "fastapi/dependencies/utils.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/dependencies/utils.py\n+++ b/fastapi/dependencies/utils.py\n@@ -871,6 +871,7 @@\n     received_body: FormData,\n ) -> Dict[str, Any]:\n     values = {}\n+    processed_keys = set()\n \n     for field in body_fields:\n         value = _get_multidict_value(field, received_body)\n@@ -902,8 +903,9 @@\n             value = serialize_sequence_value(field=field, value=results)\n         if value is not None:\n             values[field.alias] = value\n+        processed_keys.add(field.alias)\n     for key, value in received_body.items():\n-        if key not in values:\n+        if key not in processed_keys:\n             values[key] = value\n     return values\n \n", "is_truncated": false}

{"status": "ok", "stdout": "Response: {'age': None}\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}

{"status": "ok", "patch_size": 826, "files_changed": 1}