← eval_submission

fastapi_14978

failed WRONG_FIX wrong_fix · 23 tool calls · 133 s · fastapi/fastapi

Task input

🔒️ Add `strict_content_type` checking for JSON requests

🔒️ Add `strict_content_type` checking for JSON requests

Tool calls (24)

#ToolArgumentsResult
1run_command
{
 "command": "grep -rn \"json\" fastapi/params.py fastapi/routing.py fastapi/requests.py | head -20"
}
{"status": "ok", "stdout": "fastapi/params.py:72: json_schema_extra: dict[str, Any] | None = None,\nfastapi/params.py:111: current_json_schema_extra = json_sche
full
{"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}
2run_command
{
 "command": "grep -rn \"Body\" fastapi/params.py"
}
{"status": "ok", "stdout": "470:class Body(FieldInfo): # type: ignore[misc]\n582:class Form(Body): # type: ignore[misc]\n", "exit_code": 0}
full
{"status": "ok", "stdout": "470:class Body(FieldInfo):  # type: ignore[misc]\n582:class Form(Body):  # type: ignore[misc]\n", "exit_code": 0}
3read_file
{
 "filepath": "fastapi/params.py",
 "start_line": 470,
 "end_line": 510
}
{"status": "ok", "filepath": "fastapi/params.py", "content": "class Body(FieldInfo): # type: ignore[misc]\n def __init__(\n self,\n default: Any = Undefined,\n
full
{"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}
4read_file
{
 "end_line": 550,
 "filepath": "fastapi/params.py",
 "start_line": 511
}
{"status": "ok", "filepath": "fastapi/params.py", "content": " ] = _Unset,\n openapi_examples: dict[str, Example] | None = None,\n deprecated: deprecated | str
full
{"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}
5run_command
{
 "command": "grep -rn \"strict_content_type\" ."
}
{"status": "error", "error_type": "CommandError", "error_message": "Command failed with exit code 1", "details": {"stdout": "", "stderr": "", "exit_code": 1}}
full
{"status": "error", "error_type": "CommandError", "error_message": "Command failed with exit code 1", "details": {"stdout": "", "stderr": "", "exit_code": 1}}
6run_command
{
 "command": "grep -rn \"json\" fastapi/routing.py | head -20"
}
{"status": "ok", "stdout": "5:import json\n48:from fastapi.encoders import jsonable_encoder\n274: dump_json: bool = False,\n290: serializer = field.serialize_js
full
{"status": "ok", "stdout": "5:import json\n48:from fastapi.encoders import jsonable_encoder\n274:    dump_json: bool = False,\n290:        serializer = field.serialize_json if dump_json else field.serialize\n302:        return jsonable_encoder(response_content)\n370:                        json_body: Any = Undefined\n373:                            json_body = await request.json()\n379:                                if subtype == \"json\" or subtype.endswith(\"+json\"):\n380:                                    json_body = await request.json()\n381:                        if json_body != Undefined:\n382:                            body = json_body\n385:        except json.JSONDecodeError as e:\n389:                        \"type\": \"json_invalid\",\n447:                # Use the fast path (dump_json) when no custom response\n451:                # json.dumps() step.\n452:                use_dump_json = response_field is not None and isinstance(\n466:                    dump_json=use_dump_json,\n468:                if use_dump_json:\n471:                        media_type=\"application/json\",\n", "exit_code": 0}
7read_file
{
 "end_line": 400,
 "filepath": "fastapi/routing.py",
 "start_line": 350
}
{"status": "ok", "filepath": "fastapi/routing.py", "content": " _extract_endpoint_context(dependant.call)\n if dependant.call\n else EndpointContext()\n )\n\n i
full
{"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     
8read_file
{
 "end_line": 350,
 "filepath": "fastapi/routing.py",
 "start_line": 300
}
{"status": "ok", "filepath": "fastapi/routing.py", "content": "\n else:\n return jsonable_encoder(response_content)\n\n\nasync def run_endpoint_function(\n *, d
full
{"status": "ok", "filepath": "fastapi/routing.py", "content": "\n    else:\n        return jsonable_encoder(response_content)\n\n\nasync def run_endpoint_function(\n    *, dependant: Dependant, values: dict[str, Any], is_coroutine: bool\n) -> Any:\n    # Only called by get_request_handler. Has been split into its own function to\n    # facilitate profiling endpoints, since inner functions are harder to profile.\n    assert dependant.call is not None, \"dependant.call must be a function\"\n\n    if is_coroutine:\n        return await dependant.call(**values)\n    else:\n        return await run_in_threadpool(dependant.call, **values)\n\n\ndef get_request_handler(\n    dependant: Dependant,\n    body_field: ModelField | None = None,\n    status_code: int | None = None,\n    response_class: type[Response] | DefaultPlaceholder = Default(JSONResponse),\n    response_field: ModelField | None = None,\n    response_model_include: IncEx | None = None,\n    response_model_exclude: IncEx | None = None,\n    response_model_by_alias: bool = True,\n    response_model_exclude_unset: bool = False,\n    response_model_exclude_defaults: bool = False,\n    response_model_exclude_none: bool = False,\n    dependency_overrides_provider: Any | None = None,\n    embed_body_fields: bool = False,\n) -> Callable[[Request], Coroutine[Any, Any, Response]]:\n    assert dependant.call is not None, \"dependant.call must be a function\"\n    is_coroutine = dependant.is_coroutine_callable\n    is_body_form = body_field and isinstance(body_field.field_info, params.Form)\n    if isinstance(response_class, DefaultPlaceholder):\n        actual_response_class: type[Response] = response_class.value\n    else:\n        actual_response_class = response_class\n\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_middlew
9read_file
{
 "end_line": 560,
 "filepath": "fastapi/params.py",
 "start_line": 470
}
{"status": "ok", "filepath": "fastapi/params.py", "content": "class Body(FieldInfo): # type: ignore[misc]\n def __init__(\n self,\n default: Any = Undefined,\n
full
{"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
10edit_file
{
 "filepath": "fastapi/params.py",
 "new_string": "        media_type: str = \\\"application/json\\\",\n        strict_content_type: bool = False,\n        alias: str | None = None,",
 "old_string": "        media_type: str = \\\"application/json\\\",\n        alias: str | None = None,"
}
{"status": "error", "error_type": "FileEditError", "error_message": "old_string not found in file (tried exact, flexible and regex matching)"}
full
{"status": "error", "error_type": "FileEditError", "error_message": "old_string not found in file (tried exact, flexible and regex matching)"}
11read_file
{
 "end_line": 485,
 "filepath": "fastapi/params.py",
 "start_line": 470
}
{"status": "ok", "filepath": "fastapi/params.py", "content": "class Body(FieldInfo): # type: ignore[misc]\n def __init__(\n self,\n default: Any = Undefined,\n
full
{"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", "start_line": 470, "end_line": 485, "total_lines": 755, "is_truncated": false}
12edit_file
{
 "filepath": "fastapi/params.py",
 "new_string": "        media_type: str = \\\"application/json\\\",\n        strict_content_type: bool = False,\n        alias: str | None = None,",
 "old_string": "        media_type: str = \\\"application/json\\\",\n        alias: str | None = None,"
}
{"status": "error", "error_type": "FileEditError", "error_message": "old_string not found in file (tried exact, flexible and regex matching)"}
full
{"status": "error", "error_type": "FileEditError", "error_message": "old_string not found in file (tried exact, flexible and regex matching)"}
13read_file
{
 "end_line": 500,
 "filepath": "fastapi/params.py",
 "start_line": 470
}
{"status": "ok", "filepath": "fastapi/params.py", "content": "class Body(FieldInfo): # type: ignore[misc]\n def __init__(\n self,\n default: Any = Undefined,\n
full
{"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", "start_line": 470, "end_line": 500, "total_lines": 755, "is_truncated": false}
14edit_file
{
 "filepath": "fastapi/params.py",
 "new_string": "        media_type: str = \"application/json\",\n        strict_content_type: bool = False,\n        alias: str | None = None,",
 "old_string": "        media_type: str = \"application/json\",\n        alias: str | None = None,"
}
{"status": "ok", "filepath": "fastapi/params.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/params.py\n+++ b/fastapi/params.py\n@@ -476,6 +4
full
{"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}
15read_file
{
 "end_line": 530,
 "filepath": "fastapi/params.py",
 "start_line": 515
}
{"status": "ok", "filepath": "fastapi/params.py", "content": " include_in_schema: bool = True,\n json_schema_extra: dict[str, Any] | None = None,\n **extra: Any
full
{"status": "ok", "filepath": "fastapi/params.py", "content": "        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", "start_line": 515, "end_line": 530, "total_lines": 756, "is_truncated": false}
16edit_file
{
 "filepath": "fastapi/params.py",
 "new_string": "        self.embed = embed\n        self.media_type = media_type\n        self.strict_content_type = strict_content_type\n        if example is not _Unset:",
 "old_string": "        self.embed = embed\n        self.media_type = media_type\n        if example is not _Unset:"
}
{"status": "ok", "filepath": "fastapi/params.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/params.py\n+++ b/fastapi/params.py\n@@ -518,6 +5
full
{"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}
17read_file
{
 "end_line": 385,
 "filepath": "fastapi/routing.py",
 "start_line": 350
}
{"status": "ok", "filepath": "fastapi/routing.py", "content": " _extract_endpoint_context(dependant.call)\n if dependant.call\n else EndpointContext()\n )\n\n i
full
{"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", "start_line": 350, "end_line": 385, "total_lines": 4643, "is_truncated": false}
18read_file
{
 "end_line": 340,
 "filepath": "fastapi/routing.py",
 "start_line": 320
}
{"status": "ok", "filepath": "fastapi/routing.py", "content": " body_field: ModelField | None = None,\n status_code: int | None = None,\n response_class: type[R
full
{"status": "ok", "filepath": "fastapi/routing.py", "content": "    body_field: ModelField | None = None,\n    status_code: int | None = None,\n    response_class: type[Response] | DefaultPlaceholder = Default(JSONResponse),\n    response_field: ModelField | None = None,\n    response_model_include: IncEx | None = None,\n    response_model_exclude: IncEx | None = None,\n    response_model_by_alias: bool = True,\n    response_model_exclude_unset: bool = False,\n    response_model_exclude_defaults: bool = False,\n    response_model_exclude_none: bool = False,\n    dependency_overrides_provider: Any | None = None,\n    embed_body_fields: bool = False,\n) -> Callable[[Request], Coroutine[Any, Any, Response]]:\n    assert dependant.call is not None, \"dependant.call must be a function\"\n    is_coroutine = dependant.is_coroutine_callable\n    is_body_form = body_field and isinstance(body_field.field_info, params.Form)\n    if isinstance(response_class, DefaultPlaceholder):\n        actual_response_class: type[Response] = response_class.value\n    else:\n        actual_response_class = response_class\n\n", "start_line": 320, "end_line": 340, "total_lines": 4643, "is_truncated": false}
19edit_file
{
 "filepath": "fastapi/routing.py",
 "new_string": "    is_body_form = body_field and isinstance(body_field.field_info, params.Form)\n    strict_content_type = (\n        body_field and\n        not is_body_form and\n        getattr(body_field.field_info, \"strict_content_type\", False)\n    )\n    if isinstance(response_class, DefaultPlaceholder):",
 "old_string": "    is_body_form = body_field and isinstance(body_field.field_info, params.Form)\n    if isinstance(response_class, DefaultPlaceholder):"
}
{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -333,6
full
{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -333,6 +333,11 @@\n     assert dependant.call is not None, \"dependant.call must be a function\"\n     is_coroutine = dependant.is_coroutine_callable\n     is_body_form = body_field and isinstance(body_field.field_info, params.Form)\n+    strict_content_type = (\n+        body_field and\n+        not is_body_form and\n+        getattr(body_field.field_info, \"strict_content_type\", False)\n+    )\n     if isinstance(response_class, DefaultPlaceholder):\n         actual_response_class: type[Response] = response_class.value\n     else:\n", "is_truncated": false}
20read_file
{
 "end_line": 385,
 "filepath": "fastapi/routing.py",
 "start_line": 350
}
{"status": "ok", "filepath": "fastapi/routing.py", "content": " \"fastapi_middleware_astack not found in request scope\"\n )\n\n # Extract endpoint context for
full
{"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            # 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", "start_line": 350, "end_line": 385, "total_lines": 4648, "is_truncated": false}
21edit_file
{
 "filepath": "fastapi/routing.py",
 "new_string": "                        content_type_value = request.headers.get(\"content-type\")\n                        if not content_type_value:\n                            if strict_content_type:\n                                raise HTTPException(\n                                    status_code=415,\n                                    detail=\"Unsupported Media Type\",\n                                )\n                            json_body = await request.json()\n                        else:",
 "old_string": "                        content_type_value = request.headers.get(\"content-type\")\n                        if not content_type_value:\n                            json_body = await request.json()\n                        else:"
}
{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -375,6
full
{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -375,6 +375,11 @@\n                         json_body: Any = Undefined\n                         content_type_value = request.headers.get(\"content-type\")\n                         if not content_type_value:\n+                            if strict_content_type:\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}
22read_file
{
 "end_line": 385,
 "filepath": "fastapi/routing.py",
 "start_line": 370
}
{"status": "ok", "filepath": "fastapi/routing.py", "content": " body = await request.form()\n file_stack.push_async_callback(body.close)\n else:\n body_bytes =
full
{"status": "ok", "filepath": "fastapi/routing.py", "content": "                    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                            if strict_content_type:\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", "start_line": 370, "end_line": 385, "total_lines": 4653, "is_truncated": false}
23edit_file
{
 "filepath": "fastapi/routing.py",
 "new_string": "                                if subtype == \"json\" or subtype.endswith(\"+json\"):\n                                    json_body = await request.json()\n                                elif strict_content_type:\n                                    raise HTTPException(\n                                        status_code=415,\n                                        detail=\"Unsupported Media Type\",\n                                    )\n                        if json_body != Undefined:",
 "old_string": "                                if subtype == \"json\" or subtype.endswith(\"+json\"):\n                                    json_body = await request.json()\n                        if json_body != Undefined:"
}
{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -388,6
full
{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -388,6 +388,11 @@\n                                 subtype = message.get_content_subtype()\n                                 if subtype == \"json\" or subtype.endswith(\"+json\"):\n                                     json_body = await request.json()\n+                                elif strict_content_type:\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}
24submit_patch
[]
{"status": "ok", "patch_size": 2888, "files_changed": 2}
full
{"status": "ok", "patch_size": 2888, "files_changed": 2}

Patch

diff --git a/fastapi/params.py b/fastapi/params.py
index 68f98708..10319db1 100644
--- a/fastapi/params.py
+++ b/fastapi/params.py
@@ -476,6 +476,7 @@ class Body(FieldInfo):  # type: ignore[misc]
         annotation: Any | None = None,
         embed: bool | None = None,
         media_type: str = "application/json",
+        strict_content_type: bool = False,
         alias: str | None = None,
         alias_priority: int | None = _Unset,
         validation_alias: str | AliasPath | AliasChoices | None = None,
@@ -517,6 +518,7 @@ class Body(FieldInfo):  # type: ignore[misc]
     ):
         self.embed = embed
         self.media_type = media_type
+        self.strict_content_type = strict_content_type
         if example is not _Unset:
             warnings.warn(
                 "`example` has been deprecated, please use `examples` instead",
diff --git a/fastapi/routing.py b/fastapi/routing.py
index 528c9629..c573e355 100644
--- a/fastapi/routing.py
+++ b/fastapi/routing.py
@@ -333,6 +333,11 @@ def get_request_handler(
     assert dependant.call is not None, "dependant.call must be a function"
     is_coroutine = dependant.is_coroutine_callable
     is_body_form = body_field and isinstance(body_field.field_info, params.Form)
+    strict_content_type = (
+        body_field and
+        not is_body_form and
+        getattr(body_field.field_info, "strict_content_type", False)
+    )
     if isinstance(response_class, DefaultPlaceholder):
         actual_response_class: type[Response] = response_class.value
     else:
@@ -370,6 +375,11 @@ def get_request_handler(
                         json_body: Any = Undefined
                         content_type_value = request.headers.get("content-type")
                         if not content_type_value:
+                            if strict_content_type:
+                                raise HTTPException(
+                                    status_code=415,
+                                    detail="Unsupported Media Type",
+                                )
                             json_body = await request.json()
                         else:
                             message = email.message.Message()
@@ -378,6 +388,11 @@ def get_request_handler(
                                 subtype = message.get_content_subtype()
                                 if subtype == "json" or subtype.endswith("+json"):
                                     json_body = await request.json()
+                                elif strict_content_type:
+                                    raise HTTPException(
+                                        status_code=415,
+                                        detail="Unsupported Media Type",
+                                    )
                         if json_body != Undefined:
                             body = json_body
                         else:

Test output

show
==================================== ERRORS ====================================
__________ ERROR collecting tests/test_strict_content_type_nested.py ___________
tests/test_strict_content_type_nested.py:8: in <module>
    inner_strict = APIRouter(prefix="/strict", strict_content_type=True)
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E   TypeError: APIRouter.__init__() got an unexpected keyword argument 'strict_content_type'
=============================== warnings summary ===============================
../../../../../../kaggle/tmp/envs/overlays/starlette-0.52.1-py3-none-any/starlette/testclient.py:45
  /kaggle/tmp/envs/overlays/starlette-0.52.1-py3-none-any/starlette/testclient.py:45: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.
    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 warning, 1 error in 0.75s