failed WRONG_FIX wrong_fix · 23 tool calls · 133 s · fastapi/fastapi
🔒️ Add `strict_content_type` checking for JSON requests 🔒️ Add `strict_content_type` checking for JSON requests
| # | Tool | Arguments | Result |
|---|---|---|---|
| 1 | run_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_schefull{"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} |
| 2 | run_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} |
| 3 | read_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} |
| 4 | read_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} |
| 5 | run_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}} |
| 6 | run_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_jsfull{"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} |
| 7 | read_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 ifull{"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 |
| 8 | read_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 *, dfull{"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 |
| 9 | read_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 |
| 10 | edit_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)"} |
| 11 | read_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} |
| 12 | edit_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)"} |
| 13 | read_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} |
| 14 | edit_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 +4full{"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} |
| 15 | read_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: Anyfull{"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} |
| 16 | edit_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 +5full{"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} |
| 17 | read_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 ifull{"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} |
| 18 | read_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[Rfull{"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} |
| 19 | edit_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,6full{"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} |
| 20 | read_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} |
| 21 | edit_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,6full{"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} |
| 22 | read_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} |
| 23 | edit_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,6full{"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} |
| 24 | submit_patch | [] | {"status": "ok", "patch_size": 2888, "files_changed": 2}full{"status": "ok", "patch_size": 2888, "files_changed": 2} |
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:
==================================== 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