resolver / refactor (squash me)

Author: Nementon

openapi_python_client/resolver/.resolved_schema.py.swp

@@ -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

openapi_python_client/resolver/.schema_resolver.py.swp

@@ -1,43 +1,51 @@
-import urllib
-from typing import Union
+import urllib.parse
+
+from .pointer import Pointer
 
 
 class Reference:
+    """ https://tools.ietf.org/html/draft-pbryan-zyp-json-ref-03 """
+
     def __init__(self, reference: str):
         self._ref = reference
+        self._parsed_ref = urllib.parse.urlparse(reference)
 
     @property
-    def remote_relative_path(self) -> Union[str, None]:
-        if self.is_remote_ref():
-            return self._ref.split("#")[0]
-        return None
+    def path(self) -> str:
+        return urllib.parse.urldefrag(self._parsed_ref.geturl()).url
 
     @property
-    def path_parent(self) -> str:
-        path = self.path
-        parts = path.split("/")
-        parts.pop()
-        return "/".join(parts)
+    def pointer(self) -> Pointer:
+        frag = self._parsed_ref.fragment
+        if self.is_url() and frag != "" and not frag.startswith("/"):
+            frag = f"/{frag}"
 
-    @property
-    def path(self) -> str:
-        d = self._ref.split("#")[-1]
-        d = urllib.parse.unquote(d)
-        d = d.replace("~1", "/")
-        return d
+        return Pointer(frag)
+
+    def is_relative(self) -> bool:
+        """ return True if reference path is a relative path """
+        return not self.is_absolute()
+
+    def is_absolute(self) -> bool:
+        """ return True is reference path is an absolute path """
+        return self._parsed_ref.netloc != ""
 
     @property
     def value(self) -> str:
         return self._ref
 
-    def is_relative_reference(self) -> bool:
-        return self.is_remote_ref() and not self.is_url_reference()
+    def is_url(self) -> bool:
+        """ return True if the reference path is pointing to an external url location """
+        return self.is_remote() and self._parsed_ref.netloc != ""
 
-    def is_url_reference(self) -> bool:
-        return self.is_remote_ref() and (self._ref.startswith("//", 0) or self._ref.startswith("http", 0))
+    def is_remote(self) -> bool:
+        """ return True if the reference pointer is pointing to a remote document """
+        return not self.is_local()
 
-    def is_remote_ref(self) -> bool:
-        return not self.is_local_ref()
+    def is_local(self) -> bool:
+        """ return True if the reference pointer is pointing to the current document """
+        return self._parsed_ref.path == ""
 
-    def is_local_ref(self) -> bool:
-        return self._ref.startswith("#", 0)
+    def is_full_document(self) -> bool:
+        """ return True if the reference pointer is pointing to the whole document content """
+        return self.pointer.parent is None

openapi_python_client/resolver/pointer.py

@@ -34,10 +34,10 @@ def _process_remote_paths(self) -> None:
         for owner, ref_key, ref_val in self._lookup_schema_references_in(self._root, "paths"):
             ref = Reference(ref_val)
 
-            if ref.is_local_ref():
+            if ref.is_local():
                 continue
 
-            remote_path = ref.remote_relative_path
+            remote_path = ref.pointer.value
             path = ref.path
 
             if remote_path not in self._refs:
@@ -62,12 +62,12 @@ def _process_remote_components(
         for parent, ref_key, ref_val in self._lookup_schema_references(target):
             ref = Reference(ref_val)
 
-            if ref.is_local_ref():
+            if ref.is_local():
                 # print('Found local reference >> {0}'.format(ref.value))
                 if depth > 0:
                     self._transform_to_local_components(owner, ref)
             else:
-                remote_path = ref.remote_relative_path
+                remote_path = ref.pointer.value
                 if remote_path not in self._refs:
                     self._errors.append("Failed to resolve remote reference > {0}".format(remote_path))
                 else:
@@ -80,20 +80,23 @@ def _transform_to_local_components(self, owner: SchemaData, ref: Reference) -> N
 
         # print('Processing remote component > {0}'.format(ref.value))
         remote_component = self._lookup_dict(owner, ref.path)
-        root_components_dir = self._lookup_dict(self._resolved_remotes_components, ref.path_parent)
-        component_name = ref.path.split("/")[-1]
+        pointer_parent = ref.pointer.parent
+
+        if pointer_parent is not None:
+            root_components_dir = self._lookup_dict(self._resolved_remotes_components, pointer_parent.value)
+            component_name = ref.path.split("/")[-1]
 
         if component_name == "SorTransparentContainer" or component_name == "sorTransparentContainer":
             print(ref.value)
 
         if remote_component is None:
             print("Weirdy relookup of >> {0}".format(ref.value))
-            assert ref.is_local_ref() and self._lookup_dict(self._resolved_remotes_components, ref.path)
+            assert ref.is_local() and self._lookup_dict(self._resolved_remotes_components, ref.path)
             return
 
         if "$ref" in remote_component:
             subref = Reference(remote_component["$ref"])
-            if not subref.is_local_ref():
+            if not subref.is_local():
                 print("Lookup remote ref >>> {0}".format(subref.value))
                 return self._process_remote_components(remote_component)
 
@@ -118,7 +121,10 @@ def _transform_to_local_components(self, owner: SchemaData, ref: Reference) -> N
 
     def _ensure_components_dir_exists(self, ref: Reference) -> None:
         cursor = self._resolved_remotes_components
-        for key in ref.path_parent.split("/"):
+        pointer_dir = ref.pointer.parent
+        assert pointer_dir is not None
+
+        for key in pointer_dir.value.split("/"):  # noqa
             if key == "":
                 continue
 

openapi_python_client/resolver/reference.py

@@ -30,7 +30,9 @@ def __init__(self, url_or_path: Union[str, Path]):
             self._root_url = url_or_path
             try:
                 self._root_url_scheme = urllib.parse.urlparse(url_or_path).scheme
-            except Exception:
+                if self._root_url_scheme not in ["http", "https"]:
+                    raise ValueError(f"Unsupported URL scheme '{self._root_url_scheme}', expecting http or https")
+            except (TypeError, AttributeError):
                 raise urllib.error.URLError(f"Coult not parse URL > {url_or_path}")
 
     def _isapath(self, url_or_path: Union[str, Path]) -> bool:
@@ -56,15 +58,15 @@ def _resolve_schema_references(
     ) -> None:
 
         for ref in self._lookup_schema_references(root):
-            if ref.is_local_ref():
+            if ref.is_local():
                 continue
 
             try:
                 path = ref.value.split("#")[0]
                 if path in external_schemas:
                     continue
 
-                if ref.is_url_reference():
+                if ref.is_url():
                     external_schemas[path] = self._fetch_url_reference(path)
                 else:
                     external_schemas[path] = self._fetch_remote_reference(path)