← eval_submission_nothink

fastapi_14349

resolved RESOLVED PASS · 9 tool calls · 67 s · fastapi/fastapi

Task input

🐛 Fix handling of JSON Schema attributes named "$ref"

Fix handling of JSON Schema attributes named "$ref"

Related to: https://github.com/fastapi/fastapi/issues/14344

Tool calls (10)

#ToolArgumentsResult
1run_command
{
 "command": "grep -rn \"\\$ref\" ."
}
{"status": "ok", "stdout": "./fastapi/_compat/v2.py:188: if \"$ref\" not in json_schema:\n./fastapi/_compat/v2.py:264: if key == \"$ref\":\n./fastapi/_compat/v2
full
{"status": "ok", "stdout": "./fastapi/_compat/v2.py:188:    if \"$ref\" not in json_schema:\n./fastapi/_compat/v2.py:264:        if key == \"$ref\":\n./fastapi/_compat/v2.py:265:            ref_name = schema[\"$ref\"].split(\"/\")[-1]\n./fastapi/_compat/v2.py:268:                new_schema[\"$ref\"] = REF_TEMPLATE.format(model=new_name)\n./fastapi/_compat/v2.py:270:                new_schema[\"$ref\"] = schema[\"$ref\"]\n./fastapi/_compat/v2.py:310:        old_name = schema[\"$ref\"].split(\"/\")[-1]\n./fastapi/openapi/models.py:103:    ref: str = Field(alias=\"$ref\")\n./fastapi/openapi/models.py:138:    ref: Optional[str] = Field(default=None, alias=\"$ref\")\n./fastapi/openapi/models.py:322:    ref: Optional[str] = Field(default=None, alias=\"$ref\")\n./fastapi/openapi/utils.py:63:            \"items\": {\"$ref\": REF_PREFIX + \"ValidationError\"},\n./fastapi/openapi/utils.py:419:                            \"schema\": {\"$ref\": REF_PREFIX + \"HTTPValidationError\"}\n./tests/test_multi_query_errors.py:82:                                        \"$ref\": \"#/components/schemas/HTTPValidationError\"\n./tests/test_multi_query_errors.py:130:                            \"items\": {\"$ref\": \"#/components/schemas/ValidationError\"},\n./tests/test_sub_callbacks.py:133:                                    \"schema\": {\"$ref\": \"#/components/schemas/Invoice\"}\n./tests/test_sub_callbacks.py:148:                                            \"$ref\": \"#/components/schemas/HTTPValidationError\"\n./tests/test_sub_callbacks.py:165:                                                        \"$ref\": \"#/components/schemas/Event\"\n./tests/test_sub_callbacks.py:182:                                                            \"$ref\": \"#/components/schemas/HTTPValidationError\"\n./tests/test_sub_callbacks.py:201:                                                        \"$ref\": \"#/components/schemas/InvoiceEvent\"\n./tests/test_sub_callbacks.py:212:                              
2read_file
{
 "filepath": "fastapi/_compat/v2.py",
 "start_line": 180,
 "end_line": 320
}
{"status": "ok", "filepath": "fastapi/_compat/v2.py", "content": " ],\n separate_input_output_schemas: bool = True,\n) -> Dict[str, Any]:\n override_mode: Union
full
{"status": "ok", "filepath": "fastapi/_compat/v2.py", "content": "    ],\n    separate_input_output_schemas: bool = True,\n) -> Dict[str, Any]:\n    override_mode: Union[Literal[\"validation\"], None] = (\n        None if separate_input_output_schemas else \"validation\"\n    )\n    # This expects that GenerateJsonSchema was already used to generate the definitions\n    json_schema = field_mapping[(field, override_mode or field.mode)]\n    if \"$ref\" not in json_schema:\n        # TODO remove when deprecating Pydantic v1\n        # Ref: https://github.com/pydantic/pydantic/blob/d61792cc42c80b13b23e3ffa74bc37ec7c77f7d1/pydantic/schema.py#L207\n        json_schema[\"title\"] = field.field_info.title or field.alias.title().replace(\n            \"_\", \" \"\n        )\n    return json_schema\n\n\ndef get_definitions(\n    *,\n    fields: Sequence[ModelField],\n    model_name_map: ModelNameMap,\n    separate_input_output_schemas: bool = True,\n) -> Tuple[\n    Dict[Tuple[ModelField, Literal[\"validation\", \"serialization\"]], JsonSchemaValue],\n    Dict[str, Dict[str, Any]],\n]:\n    schema_generator = GenerateJsonSchema(ref_template=REF_TEMPLATE)\n    override_mode: Union[Literal[\"validation\"], None] = (\n        None if separate_input_output_schemas else \"validation\"\n    )\n    validation_fields = [field for field in fields if field.mode == \"validation\"]\n    serialization_fields = [field for field in fields if field.mode == \"serialization\"]\n    flat_validation_models = get_flat_models_from_fields(\n        validation_fields, known_models=set()\n    )\n    flat_serialization_models = get_flat_models_from_fields(\n        serialization_fields, known_models=set()\n    )\n    flat_validation_model_fields = [\n        ModelField(\n            field_info=FieldInfo(annotation=model),\n            name=model.__name__,\n            mode=\"validation\",\n        )\n        for model in flat_validation_models\n    ]\n    flat_serialization_model_fields = [\n        
3read_file
{
 "end_line": 350,
 "filepath": "fastapi/_compat/v2.py",
 "start_line": 321
}
{"status": "ok", "filepath": "fastapi/_compat/v2.py", "content": " old_name_to_new_name_map=old_name_to_new_name_map,\n )\n new_field_mapping[field_key] = new_s
full
{"status": "ok", "filepath": "fastapi/_compat/v2.py", "content": "            old_name_to_new_name_map=old_name_to_new_name_map,\n        )\n        new_field_mapping[field_key] = new_schema\n\n    new_definitions = {}\n    for key, value in definitions.items():\n        if key in old_name_to_new_name_map:\n            new_key = old_name_to_new_name_map[key]\n        else:\n            new_key = key\n        new_value = _replace_refs(\n            schema=value,\n            old_name_to_new_name_map=old_name_to_new_name_map,\n        )\n        new_definitions[new_key] = new_value\n    return new_field_mapping, new_definitions\n\n\ndef is_scalar_field(field: ModelField) -> bool:\n    from fastapi import params\n\n    return shared.field_annotation_is_scalar(\n        field.field_info.annotation\n    ) and not isinstance(field.field_info, params.Body)\n\n\ndef is_sequence_field(field: ModelField) -> bool:\n    return shared.field_annotation_is_sequence(field.field_info.annotation)\n\n\n", "start_line": 321, "end_line": 350, "total_lines": 479, "is_truncated": false}
4read_file
{
 "end_line": 275,
 "filepath": "fastapi/_compat/v2.py",
 "start_line": 250
}
{"status": "ok", "filepath": "fastapi/_compat/v2.py", "content": " model_name_map=model_name_map,\n definitions=definitions, # type: ignore[arg-type]\n field_ma
full
{"status": "ok", "filepath": "fastapi/_compat/v2.py", "content": "        model_name_map=model_name_map,\n        definitions=definitions,  # type: ignore[arg-type]\n        field_mapping=field_mapping,\n    )\n    return new_mapping, new_definitions\n\n\ndef _replace_refs(\n    *,\n    schema: Dict[str, Any],\n    old_name_to_new_name_map: Dict[str, str],\n) -> Dict[str, Any]:\n    new_schema = deepcopy(schema)\n    for key, value in new_schema.items():\n        if key == \"$ref\":\n            ref_name = schema[\"$ref\"].split(\"/\")[-1]\n            if ref_name in old_name_to_new_name_map:\n                new_name = old_name_to_new_name_map[ref_name]\n                new_schema[\"$ref\"] = REF_TEMPLATE.format(model=new_name)\n            else:\n                new_schema[\"$ref\"] = schema[\"$ref\"]\n            continue\n        if isinstance(value, dict):\n            new_schema[key] = _replace_refs(\n                schema=value,\n                old_name_to_new_name_map=old_name_to_new_name_map,\n", "start_line": 250, "end_line": 275, "total_lines": 479, "is_truncated": false}
5run_command
{
 "command": "python3 -c \"\nfrom typing import Dict, Any\nfrom copy import deepcopy\n\ndef _replace_refs(schema, old_name_to_new_name_map):\n    new_schema = deepcopy(schema)\n    for key, value in new_schema.items():\n        if key == '$ref':\n            ref_name = schema['$ref'].split('/')[-1]\n            if ref_name in old_name_to_new_name_map:\n                new_name = old_name_to_new_name_map[ref_name]\n                new_schema['$ref'] = f'#/components/schemas/{new_name}'\n            else:\n                new_schema['$ref'] = schema['$ref']\n            continue\n        if isinstance(value, dict):\n            new_schema[key] = _replace_refs(value, old_name_to_new_name_map)\n        elif isinstance(value, list):\n            new_value = []\n            for item in value:\n                if isinstance(item, dict):\n                    new_item = _replace_refs(item, old_name_to_new_name_map)\n                    new_value.append(new_item)\n                else:\n                    new_value.append(item)\n            new_schema[key] = new_value\n    return new_schema\n\n# Case 1: Normal ref\nschema1 = {'$ref': '#/components/schemas/OldModel'}\nmap1 = {'OldModel': 'NewModel'}\nprint(f'Case 1: {_replace_refs(schema1, map1)}')\n\n# Case 2: Attribute named $ref\nschema2 = {'properties': {'$ref': {'type': 'string'}}}\nmap2 = {'OldModel': 'NewModel'}\nprint(f'Case 2: {_replace_refs(schema2, map2)}')\n\n# Case 3: Nested ref in attribute named $ref\nschema3 = {'properties': {'$ref': {'$ref': '#/components/schemas/OldModel'}}}\nmap3 = {'OldModel': 'NewModel'}\nprint(f'Case 3: {_replace_refs(schema3, map3)}')\n\""
}
{"status": "error", "error_type": "CommandError", "error_message": "Command failed with exit code 1", "details": {"stdout": "Case 1: {'': '#/components/schemas/
full
{"status": "error", "error_type": "CommandError", "error_message": "Command failed with exit code 1", "details": {"stdout": "Case 1: {'': '#/components/schemas/NewModel'}\n", "stderr": "Traceback (most recent call last):\n  File \"<string>\", line 37, in <module>\n    print(f'Case 2: {_replace_refs(schema2, map2)}')\n                     ~~~~~~~~~~~~~^^^^^^^^^^^^^^^\n  File \"<string>\", line 17, in _replace_refs\n    new_schema[key] = _replace_refs(value, old_name_to_new_name_map)\n                      ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"<string>\", line 9, in _replace_refs\n    ref_name = schema[''].split('/')[-1]\n               ^^^^^^^^^^^^^^^^\nAttributeError: 'dict' object has no attribute 'split'\n", "exit_code": 1}}
6run_command
{
 "command": "python3 -c \"\nfrom typing import Dict, Any\nfrom copy import deepcopy\n\ndef _replace_refs(schema, old_name_to_new_name_map):\n    new_schema = deepcopy(schema)\n    for key, value in new_schema.items():\n        if key == '$ref':\n            # The bug is here: it uses schema['$ref'] instead of new_schema[key] or value\n            # But wait, the original code was:\n            # if key == '$ref':\n            #     ref_name = schema['$ref'].split('/')[-1]\n            #     ...\n            #     new_schema['$ref'] = ...\n            #     continue\n            \n            # If value is not a string, it's not a JSON reference, it's just a property named '$ref'\n            if not isinstance(value, str):\n                continue\n                \n            ref_name = value.split('/')[-1]\n            if ref_name in old_name_to_new_name_map:\n                new_name = old_name_to_new_name_map[ref_name]\n                new_schema[key] = f'#/components/schemas/{new_name}'\n            else:\n                new_schema[key] = value\n            continue\n        if isinstance(value, dict):\n            new_schema[key] = _replace_refs(value, old_name_to_new_name_map)\n        elif isinstance(value, list):\n            new_value = []\n            for item in value:\n                if isinstance(item, dict):\n                    new_item = _replace_refs(item, old_name_to_new_name_map)\n                    new_value.append(new_item)\n                else:\n                    new_value.append(item)\n            new_schema[key] = new_value\n    return new_schema\n\n# Case 1: Normal ref\nschema1 = {'$ref': '#/components/schemas/OldModel'}\nmap1 = {'OldModel': 'NewModel'}\nprint(f'Case 1: {_replace_refs(schema1, map1)}')\n\n# Case 2: Attribute named $ref\nschema2 = {'properties': {'$ref': {'type': 'string'}}}\nmap2 = {'OldModel': 'NewModel'}\nprint(f'Case 2: {_replace_refs(schema2, map2)}')\n\n# Case 3: Nested ref in attribute named $ref\nschema3 = {'properties': {'$ref': {'$ref': '#/components/schemas/OldModel'}}}\nmap3 = {'OldModel': 'NewModel'}\nprint(f'Case 3: {_replace_refs(schema3, map3)}')\n\""
}
{"status": "ok", "stdout": "Case 1: {'': '#/components/schemas/NewModel'}\nCase 2: {'properties': {'': {'type': 'string'}}}\nCase 3: {'properties': {'': {'': '#
full
{"status": "ok", "stdout": "Case 1: {'': '#/components/schemas/NewModel'}\nCase 2: {'properties': {'': {'type': 'string'}}}\nCase 3: {'properties': {'': {'': '#/components/schemas/OldModel'}}}\n", "exit_code": 0}
7run_command
{
 "command": "python3 -c \"\nfrom typing import Dict, Any\nfrom copy import deepcopy\n\ndef _replace_refs(schema, old_name_to_new_name_map):\n    new_schema = deepcopy(schema)\n    for key, value in new_schema.items():\n        if key == '$ref':\n            if not isinstance(value, str):\n                # It's a property named '$ref', not a JSON reference.\n                # We still need to recurse into it if it's a dict.\n                if isinstance(value, dict):\n                    new_schema[key] = _replace_refs(value, old_name_to_new_name_map)\n                elif isinstance(value, list):\n                    new_value = []\n                    for item in value:\n                        if isinstance(item, dict):\n                            new_item = _replace_refs(item, old_name_to_new_name_map)\n                            new_value.append(new_item)\n                        else:\n                            new_value.append(item)\n                    new_schema[key] = new_value\n                continue\n                \n            ref_name = value.split('/')[-1]\n            if ref_name in old_name_to_new_name_map:\n                new_name = old_name_to_new_name_map[ref_name]\n                new_schema[key] = f'#/components/schemas/{new_name}'\n            else:\n                new_schema[key] = value\n            continue\n        if isinstance(value, dict):\n            new_schema[key] = _replace_refs(value, old_name_to_new_name_map)\n        elif isinstance(value, list):\n            new_value = []\n            for item in value:\n                if isinstance(item, dict):\n                    new_item = _replace_refs(item, old_name_to_new_name_map)\n                    new_value.append(new_item)\n                else:\n                    new_value.append(item)\n            new_schema[key] = new_value\n    return new_schema\n\n# Case 1: Normal ref\nschema1 = {'$ref': '#/components/schemas/OldModel'}\nmap1 = {'OldModel': 'NewModel'}\nprint(f'Case 1: {_replace_refs(schema1, map1)}')\n\n# Case 2: Attribute named $ref\nschema2 = {'properties': {'$ref': {'type': 'string'}}}\nmap2 = {'OldModel': 'NewModel'}\nprint(f'Case 2: {_replace_refs(schema2, map2)}')\n\n# Case 3: Nested ref in attribute named $ref\nschema3 = {'properties': {'$ref': {'$ref': '#/components/schemas/OldModel'}}}\nmap3 = {'OldModel': 'NewModel'}\nprint(f'Case 3: {_replace_refs(schema3, map3)}')\n\""
}
{"status": "ok", "stdout": "Case 1: {'': '#/components/schemas/NewModel'}\nCase 2: {'properties': {'': {'type': 'string'}}}\nCase 3: {'properties': {'': {'': '#
full
{"status": "ok", "stdout": "Case 1: {'': '#/components/schemas/NewModel'}\nCase 2: {'properties': {'': {'type': 'string'}}}\nCase 3: {'properties': {'': {'': '#/components/schemas/NewModel'}}}\n", "exit_code": 0}
8run_command
{
 "command": "python3 -c \"\nfrom typing import Dict, Any\nfrom copy import deepcopy\n\ndef _replace_refs(schema, old_name_to_new_name_map):\n    new_schema = deepcopy(schema)\n    for key, value in new_schema.items():\n        if key == '$ref' and isinstance(value, str):\n            ref_name = value.split('/')[-1]\n            if ref_name in old_name_to_new_name_map:\n                new_name = old_name_to_new_name_map[ref_name]\n                new_schema[key] = f'#/components/schemas/{new_name}'\n            else:\n                new_schema[key] = value\n            continue\n        if isinstance(value, dict):\n            new_schema[key] = _replace_refs(value, old_name_to_new_name_map)\n        elif isinstance(value, list):\n            new_value = []\n            for item in value:\n                if isinstance(item, dict):\n                    new_item = _replace_refs(item, old_name_to_new_name_map)\n                    new_value.append(new_item)\n                else:\n                    new_value.append(item)\n            new_schema[key] = new_value\n    return new_schema\n\n# Case 1: Normal ref\nschema1 = {'$ref': '#/components/schemas/OldModel'}\nmap1 = {'OldModel': 'NewModel'}\nprint(f'Case 1: {_replace_refs(schema1, map1)}')\n\n# Case 2: Attribute named $ref\nschema2 = {'properties': {'$ref': {'type': 'string'}}}\nmap2 = {'OldModel': 'NewModel'}\nprint(f'Case 2: {_replace_refs(schema2, map2)}')\n\n# Case 3: Nested ref in attribute named $ref\nschema3 = {'properties': {'$ref': {'$ref': '#/components/schemas/OldModel'}}}\nmap3 = {'OldModel': 'NewModel'}\nprint(f'Case 3: {_replace_refs(schema3, map3)}')\n\""
}
{"status": "ok", "stdout": "Case 1: {'': '#/components/schemas/NewModel'}\nCase 2: {'properties': {'': {'type': 'string'}}}\nCase 3: {'properties': {'': {'': '#
full
{"status": "ok", "stdout": "Case 1: {'': '#/components/schemas/NewModel'}\nCase 2: {'properties': {'': {'type': 'string'}}}\nCase 3: {'properties': {'': {'': '#/components/schemas/NewModel'}}}\n", "exit_code": 0}
9edit_file
{
 "filepath": "fastapi/_compat/v2.py",
 "new_string": "    for key, value in new_schema.items():\n        if key == \"$ref\" and isinstance(value, str):\n            ref_name = value.split(\"/\")[-1]\n            if ref_name in old_name_to_new_name_map:\n                new_name = old_name_to_new_name_map[ref_name]\n                new_schema[\"$ref\"] = REF_TEMPLATE.format(model=new_name)\n            else:\n                new_schema[\"$ref\"] = value\n            continue\n        if isinstance(value, dict):",
 "old_string": "    for key, value in new_schema.items():\n        if key == \"$ref\":\n            ref_name = schema[\"$ref\"].split(\"/\")[-1]\n            if ref_name in old_name_to_new_name_map:\n                new_name = old_name_to_new_name_map[ref_name]\n                new_schema[\"$ref\"] = REF_TEMPLATE.format(model=new_name)\n            else:\n                new_schema[\"$ref\"] = schema[\"$ref\"]\n            continue\n        if isinstance(value, dict):"
}
{"status": "ok", "filepath": "fastapi/_compat/v2.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/_compat/v2.py\n+++ b/fastapi/_compat/v2.py\n
full
{"status": "ok", "filepath": "fastapi/_compat/v2.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/_compat/v2.py\n+++ b/fastapi/_compat/v2.py\n@@ -261,13 +261,13 @@\n ) -> Dict[str, Any]:\n     new_schema = deepcopy(schema)\n     for key, value in new_schema.items():\n-        if key == \"$ref\":\n-            ref_name = schema[\"$ref\"].split(\"/\")[-1]\n+        if key == \"$ref\" and isinstance(value, str):\n+            ref_name = value.split(\"/\")[-1]\n             if ref_name in old_name_to_new_name_map:\n                 new_name = old_name_to_new_name_map[ref_name]\n                 new_schema[\"$ref\"] = REF_TEMPLATE.format(model=new_name)\n             else:\n-                new_schema[\"$ref\"] = schema[\"$ref\"]\n+                new_schema[\"$ref\"] = value\n             continue\n         if isinstance(value, dict):\n             new_schema[key] = _replace_refs(\n", "is_truncated": false}
10submit_patch
[]
{"status": "ok", "patch_size": 880, "files_changed": 1}
full
{"status": "ok", "patch_size": 880, "files_changed": 1}

Patch

diff --git a/fastapi/_compat/v2.py b/fastapi/_compat/v2.py
index 6a87b9ae..5abb92b9 100644
--- a/fastapi/_compat/v2.py
+++ b/fastapi/_compat/v2.py
@@ -261,13 +261,13 @@ def _replace_refs(
 ) -> Dict[str, Any]:
     new_schema = deepcopy(schema)
     for key, value in new_schema.items():
-        if key == "$ref":
-            ref_name = schema["$ref"].split("/")[-1]
+        if key == "$ref" and isinstance(value, str):
+            ref_name = value.split("/")[-1]
             if ref_name in old_name_to_new_name_map:
                 new_name = old_name_to_new_name_map[ref_name]
                 new_schema["$ref"] = REF_TEMPLATE.format(model=new_name)
             else:
-                new_schema["$ref"] = schema["$ref"]
+                new_schema["$ref"] = value
             continue
         if isinstance(value, dict):
             new_schema[key] = _replace_refs(

Test output

show
..                                                                       [100%]
=============================== warnings summary ===============================
../../../../../../kaggle/tmp/envs/overlays/starlette-0.49.3-py3-none-any/starlette/testclient.py:51
  /kaggle/tmp/envs/overlays/starlette-0.49.3-py3-none-any/starlette/testclient.py:51: 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
2 passed, 1 warning in 0.74s