parser / properties / resolve local $ref

Author: Nementon

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Additions
 
+- Add support for properties local reference ($ref) resolution
 - New `--meta` command line option for specifying what type of metadata should be generated:
   - `poetry` is the default value, same behavior you're used to in previous versions
   - `setup` will generate a pyproject.toml with no Poetry information, and instead create a `setup.py` with the

openapi_python_client/parser/properties/__init__.py

@@ -519,12 +519,83 @@ def update_schemas_with_data(name: str, data: oai.Schema, schemas: Schemas) -> U
         return schemas
 
 
+def resolve_reference_and_update_schemas(
+    name: str, data: oai.Reference, schemas: Schemas, references_by_name: Dict[str, oai.Reference]
+) -> Union[Schemas, PropertyError]:
+    if _is_local_reference(data):
+        return _resolve_local_reference_schema(name, data, schemas, references_by_name)
+    else:
+        return _resolve_remote_reference_schema(name, data, schemas, references_by_name)
+
+
+def _resolve_local_reference_schema(
+    name: str, data: oai.Reference, schemas: Schemas, references_by_name: Dict[str, oai.Reference]
+) -> Union[Schemas, PropertyError]:
+    resolved_model_or_enum = _resolve_model_or_enum_reference(name, data, schemas, references_by_name)
+
+    if resolved_model_or_enum:
+        model_name = utils.pascal_case(name)
+
+        if isinstance(resolved_model_or_enum, EnumProperty):
+            schemas.enums[model_name] = resolved_model_or_enum
+
+        elif isinstance(resolved_model_or_enum, ModelProperty):
+            schemas.models[model_name] = resolved_model_or_enum
+
+        return schemas
+    else:
+        return PropertyError(data=data, detail="Failed to resolve local reference schemas.")
+
+
+def _resolve_model_or_enum_reference(
+    name: str, data: oai.Reference, schemas: Schemas, references_by_name: Dict[str, oai.Reference]
+) -> Union[EnumProperty, ModelProperty, None]:
+    target_model = _reference_model_name(data)
+    target_name = _reference_name(data)
+
+    if target_model == name or target_name == name:
+        return None  # Avoid infinite loop
+
+    if target_name in references_by_name:
+        return _resolve_model_or_enum_reference(
+            target_name, references_by_name[target_name], schemas, references_by_name
+        )
+
+    if target_model in schemas.enums:
+        return schemas.enums[target_model]
+    elif target_model in schemas.models:
+        return schemas.models[target_model]
+
+    return None
+
+
+def _resolve_remote_reference_schema(
+    name: str, data: oai.Reference, schemas: Schemas, references_by_name: Dict[str, oai.Reference]
+) -> Union[Schemas, PropertyError]:
+    return PropertyError(data=data, detail="Remote reference schemas are not supported.")
+
+
+def _is_local_reference(reference: oai.Reference) -> bool:
+    return reference.ref.startswith("#", 0)
+
+
+def _reference_model_name(reference: oai.Reference) -> str:
+    return utils.pascal_case(_reference_name(reference))
+
+
+def _reference_name(reference: oai.Reference) -> str:
+    parts = reference.ref.split("/")
+    return parts[-1]
+
+
 def build_schemas(*, components: Dict[str, Union[oai.Reference, oai.Schema]]) -> Schemas:
     """ Get a list of Schemas from an OpenAPI dict """
     schemas = Schemas()
     to_process: Iterable[Tuple[str, Union[oai.Reference, oai.Schema]]] = components.items()
     processing = True
     errors: List[PropertyError] = []
+    references_by_name: Dict[str, oai.Reference] = dict()
+    references_to_process: List[Tuple[str, oai.Reference]] = list()
 
     # References could have forward References so keep going as long as we are making progress
     while processing:
@@ -534,16 +605,26 @@ def build_schemas(*, components: Dict[str, Union[oai.Reference, oai.Schema]]) ->
         # Only accumulate errors from the last round, since we might fix some along the way
         for name, data in to_process:
             if isinstance(data, oai.Reference):
-                schemas.errors.append(PropertyError(data=data, detail="Reference schemas are not supported."))
+                references_by_name[name] = data
+                references_to_process.append((name, data))
                 continue
+
             schemas_or_err = update_schemas_with_data(name, data, schemas)
+
             if isinstance(schemas_or_err, PropertyError):
                 next_round.append((name, data))
                 errors.append(schemas_or_err)
             else:
                 schemas = schemas_or_err
-                processing = True  # We made some progress this round, do another after it's done
+                processing = True  # We made some progress this round, do another after it's donea
+
         to_process = next_round
-    schemas.errors.extend(errors)
 
+        for name, reference in references_to_process:
+            schemas_or_err = resolve_reference_and_update_schemas(name, reference, schemas, references_by_name)
+
+            if isinstance(schemas_or_err, PropertyError):
+                errors.append(schemas_or_err)
+
+    schemas.errors.extend(errors)
     return schemas

tests/test_parser/test_properties/test_init.py

@@ -1020,13 +1020,43 @@ def test_build_schemas(mocker):
     assert result.errors == [error]
 
 
-def test_build_parse_error_on_reference():
+def test_build_parse_error_on_unknown_local_reference():
     from openapi_python_client.parser.openapi import build_schemas
 
-    ref_schema = oai.Reference.construct()
+    ref_schema = oai.Reference.construct(ref="#/foobar")
     in_data = {"1": ref_schema}
     result = build_schemas(components=in_data)
-    assert result.errors[0] == PropertyError(data=ref_schema, detail="Reference schemas are not supported.")
+    assert result.errors[0] == PropertyError(data=ref_schema, detail="Failed to resolve local reference schemas.")
+
+
+def test_build_parse_success_on_known_local_reference(mocker):
+    from openapi_python_client.parser.openapi import build_schemas
+
+    build_model_property = mocker.patch(f"{MODULE_NAME}.build_model_property")
+    schemas = mocker.MagicMock()
+    build_enum_property = mocker.patch(f"{MODULE_NAME}.build_enum_property", return_value=(mocker.MagicMock(), schemas))
+    in_data = {"1": oai.Reference.construct(ref="#/foobar"), "foobar": mocker.MagicMock(enum=["val1", "val2", "val3"])}
+
+    result = build_schemas(components=in_data)
+
+    assert len(result.errors) == 0
+    assert result.enums["1"] == result.enums["foobar"]
+
+
+def test_build_parse_error_on_remote_reference():
+    from openapi_python_client.parser.openapi import build_schemas
+
+    ref_schemas = [
+        oai.Reference.construct(ref="http://foobar/../foobar.yaml#/foobar"),
+        oai.Reference.construct(ref="https://foobar/foobar.yaml#/foobar"),
+        oai.Reference.construct(ref="../foobar.yaml#/foobar"),
+        oai.Reference.construct(ref="foobar.yaml#/foobar"),
+        oai.Reference.construct(ref="//foobar#/foobar"),
+    ]
+    for ref_schema in ref_schemas:
+        in_data = {"1": ref_schema}
+        result = build_schemas(components=in_data)
+        assert result.errors[0] == PropertyError(data=ref_schema, detail="Remote reference schemas are not supported.")
 
 
 def test_build_enums(mocker):
@@ -1191,3 +1221,171 @@ def test_build_enum_property_bad_default():
 
     assert schemas == schemas
     assert err == PropertyError(detail="B is an invalid default for enum Existing", data=data)
+
+
+def test__is_local_reference():
+    from openapi_python_client.parser.properties import _is_local_reference
+
+    data_set = [
+        ("//foobar#foobar", False),
+        ("foobar#/foobar", False),
+        ("foobar.json", False),
+        ("foobar.yaml", False),
+        ("../foo/bar.json#/foobar", False),
+        ("#/foobar", True),
+        ("#/foo/bar", True),
+    ]
+
+    for data, expected_result in data_set:
+        ref = oai.Reference.construct(ref=data)
+        assert _is_local_reference(ref) == expected_result
+
+
+def test__reference_name():
+    from openapi_python_client.parser.properties import _reference_name
+
+    data_set = [
+        ("#/foobar", "foobar"),
+        ("#/foo/bar", "bar"),
+    ]
+
+    for data, expected_result in data_set:
+        ref = oai.Reference.construct(ref=data)
+        assert _reference_name(ref) == expected_result
+
+
+def test__reference_model_name():
+    from openapi_python_client.parser.properties import _reference_model_name
+
+    data_set = [
+        ("#/foobar", "Foobar"),
+        ("#/foo/bar", "Bar"),
+    ]
+
+    for data, expected_result in data_set:
+        ref = oai.Reference.construct(ref=data)
+        assert _reference_model_name(ref) == expected_result
+
+
+def test__resolve_model_or_enum_reference(mocker):
+    from openapi_python_client.parser.properties import _resolve_model_or_enum_reference
+    from openapi_python_client.parser.properties.schemas import Schemas
+
+    references_by_name = {
+        "FooBarReferenceLoop": oai.Reference.construct(ref="#/foobar"),
+        "FooBarDeeperReferenceLoop": oai.Reference.construct(ref="#/FooBarReferenceLoop"),
+        "BarFooReferenceLoop": oai.Reference.construct(ref="#/barfoo"),
+        "BarFooDeeperReferenceLoop": oai.Reference.construct(ref="#/BarFooReferenceLoop"),
+        "InfiniteReferenceLoop": oai.Reference.construct(ref="#/InfiniteReferenceLoop"),
+        "UnknownReference": oai.Reference.construct(ref="#/unknown"),
+    }
+    schemas = Schemas(enums={"Foobar": 1}, models={"Barfoo": 2})
+
+    res_1 = _resolve_model_or_enum_reference(
+        "FooBarReferenceLoop", references_by_name["FooBarReferenceLoop"], schemas, references_by_name
+    )
+    res_2 = _resolve_model_or_enum_reference(
+        "FooBarDeeperReferenceLoop", references_by_name["FooBarDeeperReferenceLoop"], schemas, references_by_name
+    )
+    res_3 = _resolve_model_or_enum_reference(
+        "BarFooReferenceLoop", references_by_name["BarFooReferenceLoop"], schemas, references_by_name
+    )
+    res_4 = _resolve_model_or_enum_reference(
+        "BarFooDeeperReferenceLoop", references_by_name["BarFooDeeperReferenceLoop"], schemas, references_by_name
+    )
+    res_5 = _resolve_model_or_enum_reference(
+        "InfiniteReferenceLoop", references_by_name["InfiniteReferenceLoop"], schemas, references_by_name
+    )
+    res_6 = _resolve_model_or_enum_reference(
+        "UnknownReference", references_by_name["UnknownReference"], schemas, references_by_name
+    )
+
+    assert res_1 == schemas.enums["Foobar"]
+    assert res_2 == schemas.enums["Foobar"]
+    assert res_3 == schemas.models["Barfoo"]
+    assert res_4 == schemas.models["Barfoo"]
+    assert res_5 == None
+    assert res_6 == None
+
+
+def test__resolve_local_reference_schema(mocker):
+    from openapi_python_client.parser.properties import _resolve_local_reference_schema
+    from openapi_python_client.parser.properties.enum_property import EnumProperty
+    from openapi_python_client.parser.properties.model_property import ModelProperty
+    from openapi_python_client.parser.properties.schemas import Schemas
+
+    references_by_name = {
+        "FooBarReferenceLoop": oai.Reference.construct(ref="#/foobar"),
+        "FooBarDeeperReferenceLoop": oai.Reference.construct(ref="#/FooBarReferenceLoop"),
+        "fooBarLowerCaseReferenceLoop": oai.Reference.construct(ref="#/foobar"),
+        "fooBarLowerCaseDeeperReferenceLoop": oai.Reference.construct(ref="#/fooBarLowerCaseReferenceLoop"),
+        "BarFooReferenceLoop": oai.Reference.construct(ref="#/barfoo"),
+        "BarFooDeeperReferenceLoop": oai.Reference.construct(ref="#/BarFooReferenceLoop"),
+        "InfiniteReferenceLoop": oai.Reference.construct(ref="#/InfiniteReferenceLoop"),
+        "UnknownReference": oai.Reference.construct(ref="#/unknown"),
+    }
+    schemas = Schemas(
+        enums={
+            "Foobar": EnumProperty(
+                name="Foobar",
+                required=False,
+                nullable=True,
+                default="foobar",
+                values=["foobar"],
+                value_type="str",
+                reference="",
+            )
+        },
+        models={
+            "Barfoo": ModelProperty(
+                name="Barfoo",
+                required=False,
+                nullable=True,
+                default="barfoo",
+                reference="",
+                required_properties=[],
+                optional_properties=[],
+                description="",
+                relative_imports=[],
+                additional_properties=[],
+            )
+        },
+    )
+
+    res_1 = _resolve_local_reference_schema(
+        "FooBarReferenceLoop", references_by_name["FooBarReferenceLoop"], schemas, references_by_name
+    )
+    res_2 = _resolve_local_reference_schema(
+        "FooBarDeeperReferenceLoop", references_by_name["FooBarDeeperReferenceLoop"], schemas, references_by_name
+    )
+    res_3 = _resolve_local_reference_schema(
+        "BarFooReferenceLoop", references_by_name["BarFooReferenceLoop"], schemas, references_by_name
+    )
+    res_4 = _resolve_local_reference_schema(
+        "BarFooDeeperReferenceLoop", references_by_name["BarFooDeeperReferenceLoop"], schemas, references_by_name
+    )
+    res_5 = _resolve_local_reference_schema(
+        "fooBarLowerCaseReferenceLoop", references_by_name["fooBarLowerCaseReferenceLoop"], schemas, references_by_name
+    )
+    res_6 = _resolve_local_reference_schema(
+        "fooBarLowerCaseDeeperReferenceLoop",
+        references_by_name["fooBarLowerCaseDeeperReferenceLoop"],
+        schemas,
+        references_by_name,
+    )
+    res_7 = _resolve_local_reference_schema(
+        "InfiniteReferenceLoop", references_by_name["InfiniteReferenceLoop"], schemas, references_by_name
+    )
+    res_8 = _resolve_local_reference_schema(
+        "UnknownReference", references_by_name["UnknownReference"], schemas, references_by_name
+    )
+
+    assert res_1 == res_2 == res_3 == res_4 == res_5 == res_6 == schemas
+    assert schemas.enums["FooBarReferenceLoop"] == schemas.enums["Foobar"]
+    assert schemas.enums["FooBarDeeperReferenceLoop"] == schemas.enums["Foobar"]
+    assert schemas.models["BarFooReferenceLoop"] == schemas.models["Barfoo"]
+    assert schemas.models["BarFooDeeperReferenceLoop"] == schemas.models["Barfoo"]
+    assert schemas.enums["FooBarLowerCaseReferenceLoop"] == schemas.enums["Foobar"]
+    assert schemas.enums["FooBarLowerCaseDeeperReferenceLoop"] == schemas.enums["Foobar"]
+    assert isinstance(res_7, PropertyError)
+    assert isinstance(res_8, PropertyError)