parser / property / add recursive reference resoltion

Author: p1-ra

openapi_python_client/parser/errors.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from enum import Enum
-from typing import Optional
+from typing import Any, Optional
 
 __all__ = ["ErrorLevel", "GeneratorError", "ParseError", "PropertyError", "ValidationError"]
 
@@ -39,5 +39,12 @@ class PropertyError(ParseError):
     header = "Problem creating a Property: "
 
 
+@dataclass
+class RecursiveReferenceInterupt(PropertyError):
+    """Error raised when a property have an recursive reference to itself"""
+
+    schemas: Optional[Any] = None  # TODO: shall not use Any here, shall be Schemas, to fix later
+
+
 class ValidationError(Exception):
     pass

openapi_python_client/parser/properties/.property.py.swp

@@ -17,12 +17,12 @@
 from ... import Config
 from ... import schema as oai
 from ... import utils
-from ..errors import ParseError, PropertyError, ValidationError
+from ..errors import ParseError, PropertyError, RecursiveReferenceInterupt, ValidationError
 from .converter import convert, convert_chain
 from .enum_property import EnumProperty
 from .model_property import ModelProperty, build_model_property
 from .property import Property
-from .schemas import Class, Schemas, parse_reference_path, update_schemas_with
+from .schemas import Class, Schemas, _Holder, _ReferencePath, parse_reference_path, update_schemas_with
 
 
 @attr.s(auto_attribs=True, frozen=True)
@@ -34,6 +34,59 @@ class NoneProperty(Property):
     template: ClassVar[Optional[str]] = "none_property.py.jinja"
 
 
+@attr.s(auto_attribs=True, frozen=True)
+class LazySelfReferenceProperty(Property):
+    """A property used to resolve recursive reference.
+    It proxyfy the required method call to its binded Property owner
+    """
+
+    owner: _Holder[Union[ModelProperty, EnumProperty, RecursiveReferenceInterupt]]
+    _resolved: bool = False
+
+    def get_base_type_string(self) -> str:
+        self._ensure_resolved()
+
+        prop = self.owner.data
+        assert isinstance(prop, Property)
+        return prop.get_base_type_string()
+
+    def get_base_json_type_string(self) -> str:
+        self._ensure_resolved()
+
+        prop = self.owner.data
+        assert isinstance(prop, Property)
+        return prop.get_base_json_type_string()
+
+    def get_type_string(self, no_optional: bool = False, json: bool = False) -> str:
+        self._ensure_resolved()
+
+        prop = self.owner.data
+        assert isinstance(prop, Property)
+        return prop.get_type_string(no_optional, json)
+
+    def get_instance_type_string(self) -> str:
+        self._ensure_resolved()
+        return super().get_instance_type_string()
+
+    def to_string(self) -> str:
+        self._ensure_resolved()
+
+        if not self.required:
+            return f"{self.python_name}: Union[Unset, {self.get_type_string()}] = UNSET"
+        else:
+            return f"{self.python_name}: {self.get_type_string()}"
+
+    def _ensure_resolved(self) -> None:
+        if self._resolved:
+            return
+
+        if not isinstance(self.owner.data, Property):
+            raise RuntimeError(f"LazySelfReferenceProperty {self.name} owner shall have been resolved.")
+        else:
+            object.__setattr__(self, "_resolved", True)
+            object.__setattr__(self, "nullable", self.owner.data.nullable)
+
+
 @attr.s(auto_attribs=True, frozen=True)
 class StringProperty(Property):
     """A property of type str"""
@@ -411,11 +464,18 @@ def _property_from_ref(
     ref_path = parse_reference_path(data.ref)
     if isinstance(ref_path, ParseError):
         return PropertyError(data=data, detail=ref_path.detail), schemas
+
     existing = schemas.classes_by_reference.get(ref_path)
-    if not existing:
+    if not existing or not existing.data:
         return PropertyError(data=data, detail="Could not find reference in parsed models or enums"), schemas
 
-    prop = attr.evolve(existing, required=required, name=name)
+    if isinstance(existing.data, RecursiveReferenceInterupt):
+        return (
+            LazySelfReferenceProperty(required=required, name=name, nullable=False, default=None, owner=existing),
+            schemas,
+        )
+
+    prop = attr.evolve(existing.data, required=required, name=name)
     if parent:
         prop = attr.evolve(prop, nullable=parent.nullable)
         if isinstance(prop, EnumProperty):
@@ -551,28 +611,44 @@ def build_schemas(
     to_process: Iterable[Tuple[str, Union[oai.Reference, oai.Schema]]] = components.items()
     still_making_progress = True
     errors: List[PropertyError] = []
-
+    recursive_references_waiting_reprocess: Dict[str, Union[oai.Reference, oai.Schema]] = dict()
+    visited: Set[_ReferencePath] = set()
+    depth = 0
     # References could have forward References so keep going as long as we are making progress
     while still_making_progress:
         still_making_progress = False
         errors = []
         next_round = []
+
         # Only accumulate errors from the last round, since we might fix some along the way
         for name, data in to_process:
             ref_path = parse_reference_path(f"#/components/schemas/{name}")
             if isinstance(ref_path, ParseError):
                 schemas.errors.append(PropertyError(detail=ref_path.detail, data=data))
                 continue
 
-            schemas_or_err = update_schemas_with(ref_path=ref_path, data=data, schemas=schemas, config=config)
+            visited.add(ref_path)
+            schemas_or_err = update_schemas_with(
+                ref_path=ref_path, data=data, schemas=schemas, visited=visited, config=config
+            )
             if isinstance(schemas_or_err, PropertyError):
-                next_round.append((name, data))
-                errors.append(schemas_or_err)
-                continue
+                if isinstance(schemas_or_err, RecursiveReferenceInterupt):
+                    up_schemas = schemas_or_err.schemas
+                    assert isinstance(up_schemas, Schemas)  # TODO fix typedef in RecursiveReferenceInterupt
+                    schemas_or_err = up_schemas
+                    recursive_references_waiting_reprocess[name] = data
+                else:
+                    next_round.append((name, data))
+                    errors.append(schemas_or_err)
+                    continue
 
             schemas = schemas_or_err
             still_making_progress = True
+        depth += 1
         to_process = next_round
 
+    if len(recursive_references_waiting_reprocess.keys()):
+        schemas = build_schemas(components=recursive_references_waiting_reprocess, schemas=schemas, config=config)
+
     schemas.errors.extend(errors)
     return schemas

openapi_python_client/parser/properties/__init__.py

@@ -11,6 +11,11 @@
 from .schemas import Class, Schemas, parse_reference_path
 
 
+@attr.s(auto_attribs=True, frozen=True)
+class RecusiveProperty(Property):
+    pass
+
+
 @attr.s(auto_attribs=True, frozen=True)
 class ModelProperty(Property):
     """A property which refers to another Schema"""
@@ -93,9 +98,12 @@ def _check_existing(prop: Property) -> Union[Property, PropertyError]:
             ref_path = parse_reference_path(sub_prop.ref)
             if isinstance(ref_path, ParseError):
                 return PropertyError(detail=ref_path.detail, data=sub_prop)
-            sub_model = schemas.classes_by_reference.get(ref_path)
-            if sub_model is None:
+
+            sub_model_ref = schemas.classes_by_reference.get(ref_path)
+            if sub_model_ref is None or not isinstance(sub_model_ref.data, Property):
                 return PropertyError(f"Reference {sub_prop.ref} not found")
+
+            sub_model = sub_model_ref.data
             if not isinstance(sub_model, ModelProperty):
                 return PropertyError("Cannot take allOf a non-object")
             for prop in chain(sub_model.required_properties, sub_model.optional_properties):

openapi_python_client/parser/properties/model_property.py

@@ -1,21 +1,21 @@
-__all__ = ["Class", "Schemas", "parse_reference_path", "update_schemas_with"]
+__all__ = ["Class", "Schemas", "parse_reference_path", "update_schemas_with", "_ReferencePath"]
 
-from typing import TYPE_CHECKING, Dict, List, NewType, Union, cast
+from typing import TYPE_CHECKING, Dict, Generic, List, NewType, Optional, Set, TypeVar, Union, cast
 from urllib.parse import urlparse
 
 import attr
 
 from ... import Config
 from ... import schema as oai
 from ... import utils
-from ..errors import ParseError, PropertyError
+from ..errors import ParseError, PropertyError, RecursiveReferenceInterupt
 
 if TYPE_CHECKING:  # pragma: no cover
     from .property import Property
 else:
     Property = "Property"
 
-
+T = TypeVar("T")
 _ReferencePath = NewType("_ReferencePath", str)
 _ClassName = NewType("_ClassName", str)
 
@@ -27,6 +27,11 @@ def parse_reference_path(ref_path_raw: str) -> Union[_ReferencePath, ParseError]
     return cast(_ReferencePath, parsed.fragment)
 
 
+@attr.s(auto_attribs=True)
+class _Holder(Generic[T]):
+    data: Optional[T]
+
+
 @attr.s(auto_attribs=True, frozen=True)
 class Class:
     """Represents Python class which will be generated from an OpenAPI schema"""
@@ -56,22 +61,33 @@ def from_string(*, string: str, config: Config) -> "Class":
 class Schemas:
     """Structure for containing all defined, shareable, and reusable schemas (attr classes and Enums)"""
 
-    classes_by_reference: Dict[_ReferencePath, Property] = attr.ib(factory=dict)
-    classes_by_name: Dict[_ClassName, Property] = attr.ib(factory=dict)
+    classes_by_reference: Dict[
+        _ReferencePath, _Holder[Union[EnumProperty, ModelProperty, RecursiveReferenceInterupt]]
+    ] = attr.ib(factory=dict)
+    classes_by_name: Dict[
+        _ClassName, _Holder[Union[EnumProperty, ModelProperty, RecursiveReferenceInterupt]]
+    ] = attr.ib(factory=dict)
     errors: List[ParseError] = attr.ib(factory=list)
 
 
 def update_schemas_with(
-    *, ref_path: _ReferencePath, data: Union[oai.Reference, oai.Schema], schemas: Schemas, config: Config
+    *,
+    ref_path: _ReferencePath,
+    data: Union[oai.Reference, oai.Schema],
+    schemas: Schemas,
+    visited: Set[_ReferencePath],
+    config: Config,
 ) -> Union[Schemas, PropertyError]:
     if isinstance(data, oai.Reference):
-        return _update_schemas_with_reference(ref_path=ref_path, data=data, schemas=schemas, config=config)
+        return _update_schemas_with_reference(
+            ref_path=ref_path, data=data, schemas=schemas, visited=visited, config=config
+        )
     else:
-        return _update_schemas_with_data(ref_path=ref_path, data=data, schemas=schemas, config=config)
+        return _update_schemas_with_data(ref_path=ref_path, data=data, schemas=schemas, visited=visited, config=config)
 
 
 def _update_schemas_with_reference(
-    *, ref_path: _ReferencePath, data: oai.Reference, schemas: Schemas, config: Config
+    *, ref_path: _ReferencePath, data: oai.Reference, schemas: Schemas, visited: Set[_ReferencePath], config: Config
 ) -> Union[Schemas, PropertyError]:
     reference_pointer = parse_reference_path(data.ref)
     if isinstance(reference_pointer, ParseError):
@@ -85,7 +101,7 @@ def _update_schemas_with_reference(
 
 
 def _update_schemas_with_data(
-    *, ref_path: _ReferencePath, data: oai.Schema, schemas: Schemas, config: Config
+    *, ref_path: _ReferencePath, data: oai.Schema, schemas: Schemas, visited: Set[_ReferencePath], config: Config
 ) -> Union[Schemas, PropertyError]:
     from . import property_from_data
 
@@ -94,8 +110,19 @@ def _update_schemas_with_data(
         data=data, name=ref_path, schemas=schemas, required=True, parent_name="", config=config
     )
 
+
+    holder = schemas.classes_by_reference.get(ref_path)
     if isinstance(prop, PropertyError):
+        if ref_path in visited and not holder:
+            holder = _Holder(data=RecursiveReferenceInterupt())
+            schemas = attr.evolve(schemas, classes_by_reference={ref_path: holder, **schemas.classes_by_reference})
+            return RecursiveReferenceInterupt(schemas=schemas)
         return prop
 
-    schemas = attr.evolve(schemas, classes_by_reference={ref_path: prop, **schemas.classes_by_reference})
+    if holder:
+        holder.data = prop
+    else:
+        schemas = attr.evolve(
+            schemas, classes_by_reference={ref_path: _Holder(data=prop), **schemas.classes_by_reference}
+        )
     return schemas