fastapi_14978

{"status": "ok", "stdout": "fastapi/params.py:72:        json_schema_extra: dict[str, Any] | None = None,\nfastapi/params.py:111:        current_json_schema_extra = json_schema_extra or extra\nfastapi/params.py:125:                \"json_schema_extra\": current_json_schema_extra,\nfastapi/params.py:183:        json_schema_extra: dict[str, Any] | None = None,\nfastapi/params.py:217:            json_schema_extra=json_schema_extra,\nfastapi/params.py:267:        json_schema_extra: dict[str, Any] | None = None,\nfastapi/params.py:299:            json_schema_extra=json_schema_extra,\nfastapi/params.py:350:        json_schema_extra: dict[str, Any] | None = None,\nfastapi/params.py:383:            json_schema_extra=json_schema_extra,\nfastapi/params.py:433:        json_schema_extra: dict[str, Any] | None = None,\nfastapi/params.py:465:            json_schema_extra=json_schema_extra,\nfastapi/params.py:478:        media_type: str = \"application/json\",\nfastapi/params.py:515:        json_schema_extra: dict[str, Any] | None = None,\nfastapi/params.py:556:        current_json_schema_extra = json_schema_extra or extra\nfastapi/params.py:569:                \"json_schema_extra\": current_json_schema_extra,\nfastapi/params.py:626:        json_schema_extra: dict[str, Any] | None = None,\nfastapi/params.py:659:            json_schema_extra=json_schema_extra,\nfastapi/params.py:708:        json_schema_extra: dict[str, Any] | None = None,\nfastapi/params.py:741:            json_schema_extra=json_schema_extra,\nfastapi/routing.py:5:import json\n", "exit_code": 0}

{"status": "ok", "stdout": "470:class Body(FieldInfo):  # type: ignore[misc]\n582:class Form(Body):  # type: ignore[misc]\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/params.py", "content": "class Body(FieldInfo):  # type: ignore[misc]\n    def __init__(\n        self,\n        default: Any = Undefined,\n        *,\n        default_factory: Callable[[], Any] | None = _Unset,\n        annotation: Any | None = None,\n        embed: bool | None = None,\n        media_type: str = \"application/json\",\n        alias: str | None = None,\n        alias_priority: int | None = _Unset,\n        validation_alias: str | AliasPath | AliasChoices | None = None,\n        serialization_alias: str | None = None,\n        title: str | None = None,\n        description: str | None = None,\n        gt: float | None = None,\n        ge: float | None = None,\n        lt: float | None = None,\n        le: float | None = None,\n        min_length: int | None = None,\n        max_length: int | None = None,\n        pattern: str | None = None,\n        regex: Annotated[\n            str | None,\n            deprecated(\n                \"Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead.\"\n            ),\n        ] = None,\n        discriminator: str | None = None,\n        strict: bool | None = _Unset,\n        multiple_of: float | None = _Unset,\n        allow_inf_nan: bool | None = _Unset,\n        max_digits: int | None = _Unset,\n        decimal_places: int | None = _Unset,\n        examples: list[Any] | None = None,\n        example: Annotated[\n            Any | None,\n            deprecated(\n                \"Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, \"\n                \"although still supported. Use examples instead.\"\n            ),\n", "start_line": 470, "end_line": 510, "total_lines": 755, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/params.py", "content": "        ] = _Unset,\n        openapi_examples: dict[str, Example] | None = None,\n        deprecated: deprecated | str | bool | None = None,\n        include_in_schema: bool = True,\n        json_schema_extra: dict[str, Any] | None = None,\n        **extra: Any,\n    ):\n        self.embed = embed\n        self.media_type = media_type\n        if example is not _Unset:\n            warnings.warn(\n                \"`example` has been deprecated, please use `examples` instead\",\n                category=FastAPIDeprecationWarning,\n                stacklevel=4,\n            )\n        self.example = example\n        self.include_in_schema = include_in_schema\n        self.openapi_examples = openapi_examples\n        kwargs = dict(\n            default=default,\n            default_factory=default_factory,\n            alias=alias,\n            title=title,\n            description=description,\n            gt=gt,\n            ge=ge,\n            lt=lt,\n            le=le,\n            min_length=min_length,\n            max_length=max_length,\n            discriminator=discriminator,\n            multiple_of=multiple_of,\n            allow_inf_nan=allow_inf_nan,\n            max_digits=max_digits,\n            decimal_places=decimal_places,\n            **extra,\n        )\n        if examples is not None:\n            kwargs[\"examples\"] = examples\n        if regex is not None:\n", "start_line": 511, "end_line": 550, "total_lines": 755, "is_truncated": false}

{"status": "error", "error_type": "CommandError", "error_message": "Command failed with exit code 1", "details": {"stdout": "", "stderr": "", "exit_code": 1}}

{"status": "ok", "stdout": "fastapi/dependencies/utils.py:968:def get_body_field(\nfastapi/routing.py:41:    get_body_field,\nfastapi/routing.py:694:        self.body_field = get_body_field(\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/dependencies/utils.py", "content": "            assert isinstance(value, sequence_types)\n            results: list[bytes | str] = []\n            for sub_value in value:\n                results.append(await sub_value.read())\n            value = serialize_sequence_value(field=field, value=results)\n        if value is not None:\n            values[get_validation_alias(field)] = value\n    field_aliases = {get_validation_alias(field) for field in body_fields}\n    for key in received_body.keys():\n        if key not in field_aliases:\n            param_values = received_body.getlist(key)\n            if len(param_values) == 1:\n                values[key] = param_values[0]\n            else:\n                values[key] = param_values\n    return values\n\n\nasync def request_body_to_args(\n    body_fields: list[ModelField],\n    received_body: dict[str, Any] | FormData | None,\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_field = body_fields[0]\n    body_to_process = received_body\n\n    fields_to_extract: list[ModelField] = body_fields\n\n    if (\n        single_not_embedded_field\n        and lenient_issubclass(first_field.field_info.annotation, BaseModel)\n        and isinstance(received_body, FormData)\n    ):\n        fields_to_extract = get_cached_model_fields(first_field.field_info.annotation)\n\n    if isinstance(received_body, FormData):\n        body_to_process = await _extract_form_body(fields_to_extract, received_body)\n\n    if single_not_embedded_field:\n        loc: tuple[str, ...] = (\"body\",)\n        v_, errors_ = _validate_value_with_model_field(\n            field=first_field, value=body_to_process, values=values, loc=loc\n       

{"status": "ok", "stdout": "fastapi/routing.py:373:                            json_body = await request.json()\nfastapi/routing.py:380:                                    json_body = await request.json()\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "            _extract_endpoint_context(dependant.call)\n            if dependant.call\n            else EndpointContext()\n        )\n\n        if dependant.path:\n            # For mounted sub-apps, include the mount path prefix\n            mount_path = request.scope.get(\"root_path\", \"\").rstrip(\"/\")\n            endpoint_ctx[\"path\"] = f\"{request.method} {mount_path}{dependant.path}\"\n\n        # Read body and auto-close files\n        try:\n            body: Any = None\n            if body_field:\n                if is_body_form:\n                    body = await request.form()\n                    file_stack.push_async_callback(body.close)\n                else:\n                    body_bytes = await request.body()\n                    if body_bytes:\n                        json_body: Any = Undefined\n                        content_type_value = request.headers.get(\"content-type\")\n                        if not content_type_value:\n                            json_body = await request.json()\n                        else:\n                            message = email.message.Message()\n                            message[\"content-type\"] = content_type_value\n                            if message.get_content_maintype() == \"application\":\n                                subtype = message.get_content_subtype()\n                                if subtype == \"json\" or subtype.endswith(\"+json\"):\n                                    json_body = await request.json()\n                        if json_body != Undefined:\n                            body = json_body\n                        else:\n                            body = body_bytes\n        except json.JSONDecodeError as e:\n            validation_error = RequestValidationError(\n                [\n                    {\n                        \"type\": \"json_invalid\",\n                        \"loc\": (\"body\", e.pos),\n     

{"status": "ok", "filepath": "fastapi/params.py", "content": "class Body(FieldInfo):  # type: ignore[misc]\n    def __init__(\n        self,\n        default: Any = Undefined,\n        *,\n        default_factory: Callable[[], Any] | None = _Unset,\n        annotation: Any | None = None,\n        embed: bool | None = None,\n        media_type: str = \"application/json\",\n        alias: str | None = None,\n        alias_priority: int | None = _Unset,\n        validation_alias: str | AliasPath | AliasChoices | None = None,\n        serialization_alias: str | None = None,\n        title: str | None = None,\n        description: str | None = None,\n        gt: float | None = None,\n        ge: float | None = None,\n        lt: float | None = None,\n        le: float | None = None,\n        min_length: int | None = None,\n        max_length: int | None = None,\n        pattern: str | None = None,\n        regex: Annotated[\n            str | None,\n            deprecated(\n                \"Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead.\"\n            ),\n        ] = None,\n        discriminator: str | None = None,\n        strict: bool | None = _Unset,\n        multiple_of: float | None = _Unset,\n        allow_inf_nan: bool | None = _Unset,\n        max_digits: int | None = _Unset,\n        decimal_places: int | None = _Unset,\n        examples: list[Any] | None = None,\n        example: Annotated[\n            Any | None,\n            deprecated(\n                \"Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, \"\n                \"although still supported. Use examples instead.\"\n            ),\n        ] = _Unset,\n        openapi_examples: dict[str, Example] | None = None,\n        deprecated: deprecated | str | bool | None = None,\n        include_in_schema: bool = True,\n        json_schema_extra: dict[str, Any] | None = None,\n        **extra: Any,\n    ):\n        self.embed = embed\n        self.media_type = media_type

{"status": "error", "error_type": "FileEditError", "error_message": "old_string not found in file (tried exact, flexible and regex matching)"}

{"status": "ok", "filepath": "fastapi/params.py", "content": "class Body(FieldInfo):  # type: ignore[misc]\n    def __init__(\n        self,\n        default: Any = Undefined,\n        *,\n        default_factory: Callable[[], Any] | None = _Unset,\n        annotation: Any | None = None,\n        embed: bool | None = None,\n        media_type: str = \"application/json\",\n        alias: str | None = None,\n        alias_priority: int | None = _Unset,\n", "start_line": 470, "end_line": 480, "total_lines": 755, "is_truncated": false}

{"status": "error", "error_type": "FileEditError", "error_message": "old_string not found in file (tried exact, flexible and regex matching)"}

{"status": "ok", "filepath": "fastapi/params.py", "content": "class Body(FieldInfo):  # type: ignore[misc]\n    def __init__(\n        self,\n        default: Any = Undefined,\n        *,\n        default_factory: Callable[[], Any] | None = _Unset,\n        annotation: Any | None = None,\n        embed: bool | None = None,\n        media_type: str = \"application/json\",\n        alias: str | None = None,\n        alias_priority: int | None = _Unset,\n        validation_alias: str | AliasPath | AliasChoices | None = None,\n        serialization_alias: str | None = None,\n        title: str | None = None,\n        description: str | None = None,\n        gt: float | None = None,\n        ge: float | None = None,\n        lt: float | None = None,\n        le: float | None = None,\n        min_length: int | None = None,\n        max_length: int | None = None,\n        pattern: str | None = None,\n        regex: Annotated[\n            str | None,\n            deprecated(\n                \"Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead.\"\n            ),\n        ] = None,\n        discriminator: str | None = None,\n        strict: bool | None = _Unset,\n        multiple_of: float | None = _Unset,\n        allow_inf_nan: bool | None = _Unset,\n        max_digits: int | None = _Unset,\n        decimal_places: int | None = _Unset,\n        examples: list[Any] | None = None,\n        example: Annotated[\n            Any | None,\n            deprecated(\n                \"Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, \"\n                \"although still supported. Use examples instead.\"\n            ),\n        ] = _Unset,\n        openapi_examples: dict[str, Example] | None = None,\n        deprecated: deprecated | str | bool | None = None,\n        include_in_schema: bool = True,\n        json_schema_extra: dict[str, Any] | None = None,\n        **extra: Any,\n    ):\n        self.embed = embed\n        self.media_type = media_type

{"status": "ok", "filepath": "fastapi/params.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/params.py\n+++ b/fastapi/params.py\n@@ -476,6 +476,7 @@\n         annotation: Any | None = None,\n         embed: bool | None = None,\n         media_type: str = \"application/json\",\n+        strict_content_type: bool = False,\n         alias: str | None = None,\n         alias_priority: int | None = _Unset,\n         validation_alias: str | AliasPath | AliasChoices | None = None,\n", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/params.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/params.py\n+++ b/fastapi/params.py\n@@ -518,6 +518,7 @@\n     ):\n         self.embed = embed\n         self.media_type = media_type\n+        self.strict_content_type = strict_content_type\n         if example is not _Unset:\n             warnings.warn(\n                 \"`example` has been deprecated, please use `examples` instead\",\n", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "        # Read body and auto-close files\n        try:\n            body: Any = None\n            if body_field:\n                if is_body_form:\n                    body = await request.form()\n                    file_stack.push_async_callback(body.close)\n                else:\n                    body_bytes = await request.body()\n                    if body_bytes:\n                        json_body: Any = Undefined\n                        content_type_value = request.headers.get(\"content-type\")\n                        if not content_type_value:\n                            json_body = await request.json()\n                        else:\n                            message = email.message.Message()\n                            message[\"content-type\"] = content_type_value\n                            if message.get_content_maintype() == \"application\":\n                                subtype = message.get_content_subtype()\n                                if subtype == \"json\" or subtype.endswith(\"+json\"):\n                                    json_body = await request.json()\n                        if json_body != Undefined:\n                            body = json_body\n                        else:\n                            body = body_bytes\n        except json.JSONDecodeError as e:\n            validation_error = RequestValidationError(\n                [\n                    {\n                        \"type\": \"json_invalid\",\n                        \"loc\": (\"body\", e.pos),\n", "start_line": 360, "end_line": 390, "total_lines": 4643, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "\n    async def app(request: Request) -> Response:\n        response: Response | None = None\n        file_stack = request.scope.get(\"fastapi_middleware_astack\")\n        assert isinstance(file_stack, AsyncExitStack), (\n            \"fastapi_middleware_astack not found in request scope\"\n        )\n\n        # Extract endpoint context for error messages\n        endpoint_ctx = (\n            _extract_endpoint_context(dependant.call)\n            if dependant.call\n            else EndpointContext()\n        )\n\n        if dependant.path:\n            # For mounted sub-apps, include the mount path prefix\n            mount_path = request.scope.get(\"root_path\", \"\").rstrip(\"/\")\n            endpoint_ctx[\"path\"] = f\"{request.method} {mount_path}{dependant.path}\"\n\n        # Read body and auto-close files\n", "start_line": 340, "end_line": 360, "total_lines": 4643, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "            \"fastapi_middleware_astack not found in request scope\"\n        )\n\n        # Extract endpoint context for error messages\n        endpoint_ctx = (\n            _extract_endpoint_context(dependant.call)\n            if dependant.call\n            else EndpointContext()\n        )\n\n        if dependant.path:\n", "start_line": 345, "end_line": 355, "total_lines": 4643, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "            _extract_endpoint_context(dependant.call)\n            if dependant.call\n            else EndpointContext()\n        )\n\n        if dependant.path:\n            # For mounted sub-apps, include the mount path prefix\n            mount_path = request.scope.get(\"root_path\", \"\").rstrip(\"/\")\n            endpoint_ctx[\"path\"] = f\"{request.method} {mount_path}{dependant.path}\"\n\n        # Read body and auto-close files\n", "start_line": 350, "end_line": 360, "total_lines": 4643, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "        # Read body and auto-close files\n        try:\n            body: Any = None\n            if body_field:\n                if is_body_form:\n                    body = await request.form()\n                    file_stack.push_async_callback(body.close)\n                else:\n                    body_bytes = await request.body()\n                    if body_bytes:\n                        json_body: Any = Undefined\n", "start_line": 360, "end_line": 370, "total_lines": 4643, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -370,6 +370,11 @@\n                         json_body: Any = Undefined\n                         content_type_value = request.headers.get(\"content-type\")\n                         if not content_type_value:\n+                            if getattr(body_field, \"strict_content_type\", False):\n+                                raise HTTPException(\n+                                    status_code=415,\n+                                    detail=\"Unsupported Media Type\",\n+                                )\n                             json_body = await request.json()\n                         else:\n                             message = email.message.Message()\n", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -383,6 +383,11 @@\n                                 subtype = message.get_content_subtype()\n                                 if subtype == \"json\" or subtype.endswith(\"+json\"):\n                                     json_body = await request.json()\n+                                elif getattr(body_field, \"strict_content_type\", False):\n+                                    raise HTTPException(\n+                                        status_code=415,\n+                                        detail=\"Unsupported Media Type\",\n+                                    )\n                         if json_body != Undefined:\n                             body = json_body\n                         else:\n", "is_truncated": false}

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