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", "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\n", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"<string>\", line 3, in <module>\n    from pydantic.fields import ModelField\nImportError: cannot import name 'ModelField' from 'pydantic.fields' (/kaggle/tmp/envs/fastapi/lib/python3.13/site-packages/pydantic/fields.py)\n", "details": {"stdout": "", "stderr": "Traceback (most recent call last):\n  File \"<string>\", line 3, in <module>\n    from pydantic.fields import ModelField\nImportError: cannot import name 'ModelField' from 'pydantic.fields' (/kaggle/tmp/envs/fastapi/lib/python3.13/site-packages/pydantic/fields.py)\n", "exit_code": 1}}

{"status": "ok", "stdout": "64:from pydantic import BaseModel\n65:from pydantic.fields import FieldInfo\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "26:    ModelField,\n174:def _get_flat_fields_from_params(fields: List[ModelField]) -> List[ModelField]:\n184:def get_flat_params(dependant: Dependant) -> List[ModelField]:\n339:    field: Optional[ModelField]\n529:def add_param_to_fields(*, field: ModelField, dependant: Dependant) -> None:\n706:    *, field: ModelField, value: Any, values: Dict[str, Any], loc: Tuple[str, ...]\n724:    field: ModelField, values: Mapping[str, Any], alias: Union[str, None] = None\n748:    fields: Sequence[ModelField],\n845:def _should_embed_body_fields(fields: List[ModelField]) -> bool:\n870:    body_fields: List[ModelField],\n912:    body_fields: List[ModelField],\n923:    fields_to_extract: List[ModelField] = body_fields\n963:) -> Optional[ModelField]:\n965:    Get a ModelField representing the request body for a path operation, combining\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "import dataclasses\nimport inspect\nfrom contextlib import AsyncExitStack, contextmanager\nfrom copy import copy, deepcopy\nfrom dataclasses import dataclass\nfrom typing import (\n    Any,\n    Callable,\n    Coroutine,\n    Dict,\n    ForwardRef,\n    List,\n    Mapping,\n    Optional,\n    Sequence,\n    Tuple,\n    Type,\n    Union,\n    cast,\n)\n\nimport anyio\nfrom fastapi import params\nfrom fastapi._compat import (\n    PYDANTIC_V2,\n    ModelField,\n    RequiredParam,\n    Undefined,\n    _is_error_wrapper,\n    _is_model_class,\n    copy_field_info,\n    create_body_model,\n    evaluate_forwardref,\n    field_annotation_is_scalar,\n    get_annotation_from_field_info,\n    get_cached_model_fields,\n    get_missing_field_error,\n    is_bytes_field,\n    is_bytes_sequence_field,\n    is_scalar_field,\n    is_scalar_sequence_field,\n    is_sequence_field,\n    is_uploadfile_or_nonable_uploadfile_annotation,\n    is_uploadfile_sequence_annotation,\n    lenient_issubclass,\n    may_v1,\n    sequence_types,\n    serialize_sequence_value,\n    value_is_sequence,\n)\n", "stderr": "", "exit_code": 0}

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

{"status": "ok", "stdout": "Case 1: result=None\nCase 2: result=None\nCase 3: result='bar'\n", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"<string>\", line 55, in <module>\n    asyncio.run(test_extract_form_body())\n    ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/root/.local/share/uv/python/cpython-3.13.14-linux-x86_64-gnu/lib/python3.13/asyncio/runners.py\", line 196, in run\n    return runner.run(main)\n           ~~~~~~~~~~^^^^^^\n  File \"/root/.local/share/uv/python/cpython-3.13.14-linux-x86_64-gnu/lib/python3.13/asyncio/runners.py\", line 119, in run\n    return self._loop.run_until_complete(task)\n           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^\n  File \"/root/.local/share/uv/python/cpython-3.13.14-linux-x86_64-gnu/lib/python3.13/asyncio/base_events.py\", line 725, in run_until_complete\n    return future.result()\n           ~~~~~~~~~~~~~^^\n  File \"<string>\", line 26, in test_extract_form_body\n    values = await _extract_form_body(body_fields, received_body)\n             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/workspace/fastapi/dependencies/utils.py\", line 885, in _extract_form_body\n    is_bytes_sequence_field(field)\n    ~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^\n  File \"/workspace/fastapi/_compat/main.py\", line 173, in is_bytes_sequence_field\n    return v2.is_bytes_sequence_field(field)  # type: ignore[arg-type]\n           ~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^\n  File \"/workspace/fastapi/_compat/v2.py\", line 360, in is_bytes_sequence_field\n    return shared.is_bytes_sequence_annotation(field.type_)\n                                               ^^^^^^^^^^^\nAttributeError: 'MockField' object has no attribute 'type_'\n", "details": {"stdout": "", "stderr": "Traceback (most recent call last):\n  File \"<string>\", line 55, in <module>\n    asyncio.run(test_extract_form_body())\n    ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/root/.local/share/uv/python/cpython-3.13.14-linux-x86_64-gnu/lib/python3.13/asyncio/runners.py\", line 196, in run\n    return r

{"status": "ok", "stdout": "Case 1: values={'foo': ''}\nCase 2: values={}\nCase 3: values={'foo': 'bar'}\n", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "CommandError", "error_message": "/bin/bash: line 1: run_command:: command not found\n/bin/bash: -c: line 42: unexpected EOF while looking for matching `\"'\n", "details": {"stdout": "", "stderr": "/bin/bash: line 1: run_command:: command not found\n/bin/bash: -c: line 42: unexpected EOF while looking for matching `\"'\n", "exit_code": 127}}

{"status": "ok", "stdout": "Case 1: values={'foo': ''}\nCase 2: values={}\nCase 3: values={'foo': 'bar'}\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "Case 1: values={'foo': ''}\nCase 2: values={}\nCase 3: values={'foo': 'bar'}\n", "stderr": "", "exit_code": 0}

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

{"status": "ok", "stdout": "Case 1: values={'foo': ''}\nCase 2: values={}\nCase 3: values={'foo': 'bar'}\n", "stderr": "", "exit_code": 0}