Remote reference support >= 0.9.0

end_to_end_tests/golden-record/my_test_api_client/models/__init__.py

@@ -2,7 +2,6 @@
 
 from .a_form_data import AFormData
 from .a_model import AModel
-from .a_model_with_properties_reference_that_are_not_object import AModelWithPropertiesReferenceThatAreNotObject
 from .a_model_with_indirect_reference_property import AModelWithIndirectReferenceProperty
 from .a_model_with_indirect_self_reference_property import AModelWithIndirectSelfReferenceProperty
 from .a_model_with_properties_reference_that_are_not_object import AModelWithPropertiesReferenceThatAreNotObject

end_to_end_tests/golden-record/my_test_api_client/models/a_model_with_indirect_reference_property.py

@@ -31,9 +31,11 @@ def to_dict(self) -> Dict[str, Any]:
     @classmethod
     def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
         d = src_dict.copy()
-        an_enum_indirect_ref: Union[Unset, AnEnum] = UNSET
         _an_enum_indirect_ref = d.pop("an_enum_indirect_ref", UNSET)
-        if not isinstance(_an_enum_indirect_ref, Unset):
+        an_enum_indirect_ref: Union[Unset, AnEnum]
+        if isinstance(_an_enum_indirect_ref, Unset):
+            an_enum_indirect_ref = UNSET
+        else:
             an_enum_indirect_ref = AnEnum(_an_enum_indirect_ref)
 
         a_model_with_indirect_reference_property = cls(

end_to_end_tests/golden-record/my_test_api_client/models/a_model_with_indirect_self_reference_property.py

@@ -44,9 +44,11 @@ def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
         d = src_dict.copy()
         required_self_ref = d.pop("required_self_ref")
 
-        an_enum: Union[Unset, AnEnum] = UNSET
         _an_enum = d.pop("an_enum", UNSET)
-        if not isinstance(_an_enum, Unset):
+        an_enum: Union[Unset, AnEnum]
+        if isinstance(_an_enum, Unset):
+            an_enum = UNSET
+        else:
             an_enum = AnEnum(_an_enum)
 
         optional_self_ref = d.pop("optional_self_ref", UNSET)

openapi_python_client/__init__.py

@@ -3,20 +3,21 @@
 import shutil
 import subprocess
 import sys
+import urllib
 from enum import Enum
 from pathlib import Path
-from typing import Any, Dict, Optional, Sequence, Union
+from typing import Any, Dict, Optional, Sequence, Union, cast
 
 import httpcore
 import httpx
-import yaml
 from jinja2 import BaseLoader, ChoiceLoader, Environment, FileSystemLoader, PackageLoader
 
 from openapi_python_client import utils
 
 from .config import Config
 from .parser import GeneratorData, import_string_from_class
 from .parser.errors import GeneratorError
+from .resolver.schema_resolver import SchemaResolver
 from .utils import snake_case
 
 if sys.version_info.minor < 8:  # version did not exist before 3.8, need to use a backport
@@ -351,20 +352,21 @@ def update_existing_client(
 
 
 def _get_document(*, url: Optional[str], path: Optional[Path]) -> Union[Dict[str, Any], GeneratorError]:
-    yaml_bytes: bytes
     if url is not None and path is not None:
         return GeneratorError(header="Provide URL or Path, not both.")
-    if url is not None:
-        try:
-            response = httpx.get(url)
-            yaml_bytes = response.content
-        except (httpx.HTTPError, httpcore.NetworkError):
-            return GeneratorError(header="Could not get OpenAPI document from provided URL")
-    elif path is not None:
-        yaml_bytes = path.read_bytes()
-    else:
+
+    if url is None and path is None:
         return GeneratorError(header="No URL or Path provided")
+
+    source = cast(Union[str, Path], (url if url is not None else path))
     try:
-        return yaml.safe_load(yaml_bytes)
-    except yaml.YAMLError:
+        resolver = SchemaResolver(source)
+        result = resolver.resolve()
+        if len(result.errors) > 0:
+            return GeneratorError(header="; ".join(result.errors))
+    except (httpx.HTTPError, httpcore.NetworkError, urllib.error.URLError):
+        return GeneratorError(header="Could not get OpenAPI document from provided URL")
+    except Exception:
         return GeneratorError(header="Invalid YAML from provided source")
+
+    return result.schema

openapi_python_client/parser/properties/schemas.py

@@ -61,12 +61,10 @@ 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, _Holder[Union[Property, RecursiveReferenceInterupt]]
-    ] = attr.ib(factory=dict)
-    classes_by_name: Dict[
-        _ClassName, _Holder[Union[Property, RecursiveReferenceInterupt]]
-    ] = attr.ib(factory=dict)
+    classes_by_reference: Dict[_ReferencePath, _Holder[Union[Property, RecursiveReferenceInterupt]]] = attr.ib(
+        factory=dict
+    )
+    classes_by_name: Dict[_ClassName, _Holder[Union[Property, RecursiveReferenceInterupt]]] = attr.ib(factory=dict)
     errors: List[ParseError] = attr.ib(factory=list)
 
 

openapi_python_client/parser/responses.py

@@ -21,6 +21,7 @@ class Response:
 
 _SOURCE_BY_CONTENT_TYPE = {
     "application/json": "response.json()",
+    "application/problem+json": "response.json()",
     "application/vnd.api+json": "response.json()",
     "application/octet-stream": "response.content",
     "text/html": "response.text",

openapi_python_client/resolver/collision_resolver.py

@@ -0,0 +1,148 @@
+import hashlib
+import re
+from typing import Any, Dict, List, Tuple
+
+from .reference import Reference
+from .resolver_types import SchemaData
+
+
+class CollisionResolver:
+    def __init__(self, root: SchemaData, refs: Dict[str, SchemaData], errors: List[str], parent: str):
+        self._root: SchemaData = root
+        self._refs: Dict[str, SchemaData] = refs
+        self._errors: List[str] = errors
+        self._parent = parent
+        self._refs_index: Dict[str, str] = dict()
+        self._schema_index: Dict[str, Reference] = dict()
+        self._keys_to_replace: Dict[str, Tuple[int, SchemaData, List[str]]] = dict()
+
+    def _browse_schema(self, attr: Any, root_attr: Any) -> None:
+        if isinstance(attr, dict):
+            attr_copy = {**attr}  # Create a shallow copy
+            for key, val in attr_copy.items():
+                if key == "$ref":
+                    ref = Reference(val, self._parent)
+                    value = ref.pointer.value
+
+                    assert value
+
+                    schema = self._get_from_ref(ref, root_attr)
+                    hashed_schema = self._reference_schema_hash(schema)
+
+                    if value in self._refs_index.keys():
+                        if self._refs_index[value] != hashed_schema:
+                            if ref.is_local():
+                                self._increment_ref(ref, root_attr, hashed_schema, attr, key)
+                            else:
+                                assert ref.abs_path in self._refs.keys()
+                                self._increment_ref(ref, self._refs[ref.abs_path], hashed_schema, attr, key)
+                    else:
+                        self._refs_index[value] = hashed_schema
+
+                    if hashed_schema in self._schema_index.keys():
+                        existing_ref = self._schema_index[hashed_schema]
+                        if (
+                            existing_ref.pointer.value != ref.pointer.value
+                            and ref.pointer.tokens()[-1] == existing_ref.pointer.tokens()[-1]
+                        ):
+                            self._errors.append(f"Found a duplicate schema in {existing_ref.value} and {ref.value}")
+                    else:
+                        self._schema_index[hashed_schema] = ref
+
+                else:
+                    self._browse_schema(val, root_attr)
+
+        elif isinstance(attr, list):
+            for val in attr:
+                self._browse_schema(val, root_attr)
+
+    def _get_from_ref(self, ref: Reference, attr: SchemaData) -> SchemaData:
+        if ref.is_remote():
+            assert ref.abs_path in self._refs.keys()
+            attr = self._refs[ref.abs_path]
+        cursor = attr
+        query_parts = ref.pointer.tokens()
+
+        for key in query_parts:
+            if key == "":
+                continue
+
+            if isinstance(cursor, dict) and key in cursor:
+                cursor = cursor[key]
+            else:
+                self._errors.append(f"Did not find data corresponding to the reference {ref.value}")
+
+        if list(cursor) == ["$ref"]:
+            ref2 = cursor["$ref"]
+            ref2 = re.sub(r"(.*)_\d", r"\1", ref2)
+            ref2 = Reference(ref2, self._parent)
+            if ref2.is_remote():
+                attr = self._refs[ref2.abs_path]
+            return self._get_from_ref(ref2, attr)
+
+        return cursor
+
+    def _increment_ref(
+        self, ref: Reference, schema: SchemaData, hashed_schema: str, attr: Dict[str, Any], key: str
+    ) -> None:
+        i = 2
+        value = ref.pointer.value
+        incremented_value = value + "_" + str(i)
+
+        while incremented_value in self._refs_index.keys():
+            if self._refs_index[incremented_value] == hashed_schema:
+                if ref.value not in self._keys_to_replace.keys():
+                    break  # have to increment target key aswell
+                else:
+                    attr[key] = ref.value + "_" + str(i)
+                    return
+            else:
+                i = i + 1
+                incremented_value = value + "_" + str(i)
+
+        attr[key] = ref.value + "_" + str(i)
+        self._refs_index[incremented_value] = hashed_schema
+        self._keys_to_replace[ref.value] = (i, schema, ref.pointer.tokens())
+
+    def _modify_root_ref_name(self, query_parts: List[str], i: int, attr: SchemaData) -> None:
+        cursor = attr
+        last_key = query_parts[-1]
+
+        for key in query_parts:
+            if key == "":
+                continue
+
+            if key == last_key and key + "_" + str(i) not in cursor:
+                assert key in cursor, "Didnt find %s in %s" % (key, attr)
+                cursor[key + "_" + str(i)] = cursor.pop(key)
+                return
+
+            if isinstance(cursor, dict) and key in cursor:
+                cursor = cursor[key]
+            else:
+                return
+
+    def resolve(self) -> None:
+        self._browse_schema(self._root, self._root)
+        for file, schema in self._refs.items():
+            self._browse_schema(schema, schema)
+        for a, b in self._keys_to_replace.items():
+            self._modify_root_ref_name(b[2], b[0], b[1])
+
+    def _reference_schema_hash(self, schema: Dict[str, Any]) -> str:
+        md5 = hashlib.md5()
+        hash_elms = []
+        for key in schema.keys():
+            if key == "description":
+                hash_elms.append(schema[key])
+            if key == "type":
+                hash_elms.append(schema[key])
+            if key == "allOf":
+                for item in schema[key]:
+                    hash_elms.append(str(item))
+
+            hash_elms.append(key)
+
+        hash_elms.sort()
+        md5.update(";".join(hash_elms).encode("utf-8"))
+        return md5.hexdigest()

openapi_python_client/resolver/data_loader.py

@@ -0,0 +1,24 @@
+import json
+
+import yaml
+
+from .resolver_types import SchemaData
+
+
+class DataLoader:
+    @classmethod
+    def load(cls, path: str, data: bytes) -> SchemaData:
+        data_type = path.split(".")[-1].casefold()
+
+        if data_type == "json":
+            return cls.load_json(data)
+        else:
+            return cls.load_yaml(data)
+
+    @classmethod
+    def load_json(cls, data: bytes) -> SchemaData:
+        return json.loads(data)
+
+    @classmethod
+    def load_yaml(cls, data: bytes) -> SchemaData:
+        return yaml.safe_load(data)

openapi_python_client/resolver/pointer.py

@@ -0,0 +1,48 @@
+import urllib.parse
+from typing import List, Union
+
+
+class Pointer:
+    """https://tools.ietf.org/html/rfc6901"""
+
+    def __init__(self, pointer: str) -> None:
+        if pointer is None or pointer != "" and not pointer.startswith("/"):
+            raise ValueError(f'Invalid pointer value {pointer}, it must match: *( "/" reference-token )')
+
+        self._pointer = pointer
+
+    @property
+    def value(self) -> str:
+        return self._pointer
+
+    @property
+    def parent(self) -> Union["Pointer", None]:
+        tokens = self.tokens(False)
+
+        if len(tokens) > 1:
+            tokens.pop()
+            return Pointer("/".join(tokens))
+        else:
+            assert tokens[-1] == ""
+            return None
+
+    def tokens(self, unescape: bool = True) -> List[str]:
+        tokens = []
+
+        if unescape:
+            for token in self._pointer.split("/"):
+                tokens.append(self._unescape(token))
+        else:
+            tokens = self._pointer.split("/")
+
+        return tokens
+
+    @property
+    def unescapated_value(self) -> str:
+        return self._unescape(self._pointer)
+
+    def _unescape(self, data: str) -> str:
+        data = urllib.parse.unquote(data)
+        data = data.replace("~1", "/")
+        data = data.replace("~0", "~")
+        return data