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": "error", "error_type": "FileReadError", "error_message": "'>' not supported between instances of 'int' and 'str'"}

{"status": "error", "error_type": "FileReadError", "error_message": "'>' not supported between instances of 'int' and 'str'"}

{"status": "ok", "stdout": "869:async def _extract_form_body(\n", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "FileReadError", "error_message": "'>' not supported between instances of 'int' and 'str'"}

{"status": "ok", "stdout": "            \"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 single field, not multiple\n    if len(body_param_names_set) > 1:\n        return True\n    first_field = fields[0]\n    # If it explicitly 

{"status": "error", "error_type": "FileReadError", "error_message": "'>' not supported between instances of 'int' and 'str'"}

{"status": "ok", "stdout": "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) == 1 and not embed_body_fields\n    first_fiel

{"status": "ok", "stdout": "723:def _get_multidict_value(\n788:        value = _get_multidict_value(field, received_params, alias=alias)\n810:        value = _get_multidict_value(field, received_params)\n876:        value = _get_multidict_value(field, received_body)\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "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:\n            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\n\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  

{"status": "ok", "stdout": "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) == 1 and not embed_body_fields\n    first_fiel

{"status": "ok", "filepath": "reproduce_issue.py", "size": 1079}

{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"/workspace/reproduce_issue.py\", line 33, in <module>\n    test_empty_string_form_value()\n    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^\n  File \"/workspace/reproduce_issue.py\", line 20, in test_empty_string_form_value\n    assert response.json() == {\"name\": None}\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "details": {"stdout": "", "stderr": "Traceback (most recent call last):\n  File \"/workspace/reproduce_issue.py\", line 33, in <module>\n    test_empty_string_form_value()\n    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^\n  File \"/workspace/reproduce_issue.py\", line 20, in test_empty_string_form_value\n    assert response.json() == {\"name\": None}\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "exit_code": 1}}

{"status": "error", "error_type": "CommandError", "error_message": "  File \"<string>\", line 1\n    from fastapi import FastAPI, Form; from fastapi.testclient import TestClient; app = FastAPI(); @app.post('/test')\\nasync def test_endpoint(name: str = Form(None)): return {'name': name}\\nclient = TestClient(app); print(client.post('/test', data={'name': ''}).json()\n                                                                                                   ^\nSyntaxError: invalid syntax\n", "details": {"stdout": "", "stderr": "  File \"<string>\", line 1\n    from fastapi import FastAPI, Form; from fastapi.testclient import TestClient; app = FastAPI(); @app.post('/test')\\nasync def test_endpoint(name: str = Form(None)): return {'name': name}\\nclient = TestClient(app); print(client.post('/test', data={'name': ''}).json()\n                                                                                                   ^\nSyntaxError: invalid syntax\n", "exit_code": 1}}

{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"/workspace/reproduce_issue.py\", line 33, in <module>\n    test_empty_string_form_value()\n    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^\n  File \"/workspace/reproduce_issue.py\", line 20, in test_empty_string_form_value\n    assert response.json() == {\"name\": None}\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "details": {"stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.13.14, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: enabled, timer_resolution: 1.0ns)\nrootdir: /workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: anyio-4.15.1, sugar-1.1.1, inline-snapshot-0.35.4, timeout-2.4.0, xdist-3.8.0, codspeed-5.0.3, cov-7.1.0\ncollected 4 items\n\nreproduce_issue.py FF..                                                  [100%]\n\n=================================== FAILURES ===================================\n________________________________ test_endpoint _________________________________\nasync def functions are not natively supported.\nYou need to install a suitable plugin for your async framework, for example:\n  - anyio\n  - pytest-asyncio\n  - pytest-tornasync\n  - pytest-trio\n  - pytest-twisted\n_________________________ test_empty_string_form_value _________________________\n\n    def test_empty_string_form_value():\n        # When an empty string is sent in a form, it should be interpreted as missing (None)\n        # if the default is None.\n        response = client.post(\"/test\", data={\"name\": \"\"})\n        assert response.status_code == 200\n>       assert response.json() == {\"name\": None}\nE       AssertionError: assert {'name': ''} == {'name': None}\nE         \nE         Differing items:\nE         {'name': ''} != {'name': None}\nE         Use -v to get more diff\n\nreproduce_i

{"status": "ok", "stdout": "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) == 1 and not embed_body_fields\n    first_fiel

{"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": "All tests passed!\n", "stderr": "", "exit_code": 0}

{"status": "ok", "patch_size": 2096, "files_changed": 2}