Start of OpenAPI to Hydra Parser v2

- Separated concerns by adding `/parsers` and `/processors`
- Parsers are responsible for parsing the text of the OpenAPI Spec and extracting relevant information for creating a `HydraDoc`
- Processors then use this information and act as an adapter to the `hydra-python-core` which then convert it to a Hydra Object
- These Hydra objects are then used by the `APIClassProcessor` to finally create a relevant Hydra Class and then send it to the  `OpenAPIDocParser` which assimilates everything to create a `HydraDoc`

.gitignore

@@ -90,6 +90,9 @@ ENV/
 env.bak/
 venv.bak/
 
+# VScode project settings
+.vscode
+
 # Spyder project settings
 .spyderproject
 .spyproject

hydra_openapi_parser/hydra_openapi_parser_v2/exceptions.py

@@ -0,0 +1,8 @@
+class HydraCollectionException(Exception):
+    def __init__(self, message) -> None:
+        super().__init__(message)
+
+
+class ExpectsValueError(Exception):
+    def __init__(self, message) -> None:
+        super().__init__(message)

hydra_openapi_parser/hydra_openapi_parser_v2/openapi_parser.py

@@ -0,0 +1,51 @@
+import os
+import yaml
+from processors.api_info_processor import APIInfoProcessor
+from processors.api_class_processor import APIClassProcessor
+from utils import gen_entrypoint, gen_doc_file
+from hydra_python_core.doc_writer import HydraStatus
+
+
+class OpenAPIDocParser:
+    def __init__(self, inp_path: str) -> None:
+        # construct the path for the input OpenAPI doc
+        self.current_dir = os.path.dirname(__file__)
+        input_file_path = os.path.join(self.current_dir, inp_path)
+        with open(input_file_path) as stream:
+            try:
+                self.openapi_doc = yaml.safe_load(stream)
+            except yaml.YAMLError as exc:
+                print(exc)
+
+    def parse(self, out_path: str) -> str:
+        # create basic hydra doc with general info (@id, @context, description etc.)
+        info = APIInfoProcessor(self.openapi_doc)
+        api_info_doc = info.generate()
+        api_doc = gen_entrypoint(api_info_doc)
+
+        # create supported classes for hydra doc
+        api_classes = APIClassProcessor(self.openapi_doc)
+        supported_classes, supported_collections = api_classes.generate()
+        for supported_class in supported_classes:
+            api_doc.add_supported_class(supported_class)
+
+        # create supported collections for hydra doc
+        for supported_collection in supported_collections:
+            api_doc.add_supported_collection(supported_collection)
+
+        hydra_doc = api_doc.generate()
+
+        if out_path:
+            # construct the path for the output Hydra doc
+            output_file_path = os.path.join(self.current_dir, out_path)
+            with open(output_file_path, "w") as f:
+                f.write(gen_doc_file(hydra_doc))
+
+        return hydra_doc
+
+
+if __name__ == "__main__":
+    petstore_doc_path = "../samples/v2/petstore-expanded.yaml"
+    uspto_doc_path = "../samples/v2/uspto.yaml"
+    openapi_doc = OpenAPIDocParser(petstore_doc_path)
+    hydra_doc = openapi_doc.parse("../samples/v2/hydra_doc_sample.py")

hydra_openapi_parser/hydra_openapi_parser_v2/parsers/api_info_parser.py

@@ -0,0 +1,37 @@
+import re
+from urllib.parse import urlparse
+
+
+class InfoParser:
+    def __init__(self, doc) -> None:
+        self.doc = doc
+
+    def parse(self):
+        info = dict()
+        info_ = dict()
+        for key, value in self.doc.get("info").items():
+            info[key] = value
+        info["url"] = self.doc.get("servers")[0].get("url")
+
+        # check for variables in the url
+        if self.doc.get("servers")[0].get("variables"):
+            server = self.doc.get("servers")[0]
+            for variable, variable_details in server.get("variables").items():
+                info["url"] = info["url"].replace(
+                    rf"{{{variable}}}", variable_details.get("default")
+                )
+        url = urlparse(info["url"])
+
+        info_ = {
+            "api": info.get(url.path.split("/")[0], "api"),
+            "title": info.get("title", ""),
+            "desc": info.get("description", ""),
+            "base_url": f"{url.scheme}://{url.netloc}",
+            "doc_name": info.get("", "vocab"),
+        }
+        info_["api"] = "{}/v{}".format(
+            info_["api"], info.get("version", "1.0.0").split(".")[0]
+        )
+        info_["entrypoint"] = f'{info_["base_url"]}/{info_["api"]}'
+        info_["id"] = f'{info_["entrypoint"]}/{info_["doc_name"]}'
+        return info_

hydra_openapi_parser/hydra_openapi_parser_v2/parsers/components_parser.py

@@ -0,0 +1,12 @@
+from utils import component_class_mapping
+
+
+class ComponentParser:
+    def __init__(self, component, definition) -> None:
+        self.component = component
+        self.definition = definition
+
+    def parse(self):
+        Parser = component_class_mapping(self.component)
+        parser = Parser(self.definition)
+        return parser.parse()

hydra_openapi_parser/hydra_openapi_parser_v2/parsers/method_parser.py

@@ -0,0 +1,64 @@
+from hydra_python_core.doc_writer import (
+    HydraClassOp,
+    HydraClassProp,
+    HydraError,
+    HydraStatus,
+)
+from exceptions import HydraCollectionException
+from processors.api_info_processor import APIInfoProcessor
+from parsers.param_parser import ParameterParser
+from parsers.resp_parser import ResponseParser
+from processors.op_processor import OperationProcessor
+from parsers.schema_parser import SchemaParser
+
+from typing import Any, List, Dict, Union
+
+
+class MethodParser:
+    def __init__(self, method: str, method_details: Dict[str, Any], id: str) -> None:
+        self.method = method.upper()
+        self.method_details = method_details
+        self.id = id
+
+    def parse(self) -> List[Union[HydraClassOp, List[HydraClassProp]]]:
+        method_title = str
+        hydra_props: List[HydraClassProp] = []
+        hydra_op: HydraClassOp
+        possible_status: List[Union[HydraStatus, HydraError]] = []
+        expects_resource = ""
+        returns_resource = ""
+        for key, value in self.method_details.items():
+            if key == "parameters":
+                for parameter in value:
+                    param_parser = ParameterParser(parameter)
+                    hydra_class_prop = param_parser.parse()
+                    hydra_props.append(hydra_class_prop)
+            elif key == "responses":
+                for code, response in value.items():
+                    response_parser = ResponseParser(code, response)
+                    hydra_status = response_parser.parse()
+                    possible_status.append(hydra_status)
+                    if response_parser.parse_code() != 500:
+                        returns_resource = response_parser.parse_returns()
+
+            elif key == "operationId":
+                method_title = value
+            elif key == "requestBody":
+                request_content = value.get("content")
+                for _, expects in request_content.items():
+                    schema_parser = SchemaParser(expects.get("schema"))
+                    hydra_classes, _ = schema_parser.parse()
+                    for title, _ in hydra_classes.items():
+                        expects_resource = title
+
+        operation_processor = OperationProcessor(
+            title=method_title,
+            method=self.method,
+            id=self.id,
+            possible_status=possible_status,
+            expects=expects_resource,
+            returns=returns_resource,
+        )
+        hydra_op = operation_processor.generate()
+
+        return [hydra_op, hydra_props]

hydra_openapi_parser/hydra_openapi_parser_v2/parsers/param_parser.py

@@ -0,0 +1,41 @@
+from typing import Type
+from processors.prop_processor import PropertyProcessor
+from utils import type_ref_mapping
+from exceptions import ExpectsValueError, HydraCollectionException
+
+
+class ParameterParser:
+    def __init__(self, parameter) -> None:
+        self.parameter = parameter
+
+    def parse(self):
+        is_collection = False
+        prop = "null"
+        title = ("null",)
+        required = (False,)
+
+        for key, value in self.parameter.items():
+            if key == "schema":
+                schema = value
+                if schema.get("type") == "array":
+                    is_collection = True
+                else:
+                    try:
+                        prop = type_ref_mapping(schema.get("type"))
+                    except KeyError:
+                        raise ExpectsValueError(
+                            "{} is incorrect parameter for 'supportedProperty'.".format(
+                                schema.get("type")
+                            )
+                        )
+            elif key == "required":
+                required = value
+            elif key == "name":
+                title = value
+
+        if is_collection:
+            raise HydraCollectionException("Parsed parameter is a collection.")
+
+        property_processor = PropertyProcessor(prop, title, required)
+        hydra_prop = property_processor.generate()
+        return hydra_prop

hydra_openapi_parser/hydra_openapi_parser_v2/parsers/path_parser.py

@@ -0,0 +1,46 @@
+from typing import Union, List, Dict
+from hydra_python_core.doc_writer import (
+    HydraClassOp,
+    HydraClassProp,
+)
+from exceptions import HydraCollectionException
+from parsers.method_parser import MethodParser
+
+
+class PathParser:
+    def __init__(self, path_name, path_method, id) -> None:
+        self.path_method = path_method
+        self.id = f'{id}?resource={path_name.split("/")[1].capitalize()}'
+        self.hydra_class_ops = []
+        self.hydra_class_props = []
+        self.hydra_collection_ops = {}
+
+    def is_parsed(self):
+        if self.hydra_class_ops or self.hydra_collection_ops or self.hydra_class_props:
+            return True
+        return False
+
+    def get_class_props(self):
+        if not self.is_parsed():
+            self.parse()
+        return self.hydra_class_props
+
+    def get_class_ops(self) -> List[HydraClassOp]:
+        if not self.is_parsed():
+            self.parse()
+        return self.hydra_class_ops
+
+    def get_collection_ops(self) -> Dict[str, bool]:
+        if not self.is_parsed():
+            self.parse()
+        return self.hydra_collection_ops
+
+    def parse(self) -> None:
+        for method_name, method_details in self.path_method.items():
+            try:
+                method_parser = MethodParser(method_name, method_details, self.id)
+                hydra_class_op_, hydra_class_props_ = method_parser.parse()
+                self.hydra_class_ops.append(hydra_class_op_)
+                self.hydra_class_props.extend(hydra_class_props_)
+            except HydraCollectionException:
+                self.hydra_collection_ops[method_name] = True

hydra_openapi_parser/hydra_openapi_parser_v2/parsers/ref_parser.py

@@ -0,0 +1,36 @@
+from utils import parser_class_mapping
+from os.path import isdir
+
+
+class RefParser:
+    def __init__(self, pointer) -> None:
+        self.pointer = pointer
+
+    def find_root(self):
+        path_to_ref = self.pointer.split("/")
+        if path_to_ref[0] == "#":
+            return path_to_ref[::1]
+        elif isdir(path_to_ref[0]):
+            return "directory"
+        else:
+            return "url"
+
+    def parse(self):
+        root = self.find_root()
+        hydra_class = {}
+        hydra_collection = {}
+        if root == "directory":
+            pass
+        elif root == "url":
+            pass
+        else:
+            # components within the same file
+            from processors.api_class_processor import APIClassProcessor
+
+            component_type = root[2]
+            if component_type == "schemas":
+                hydra_entity = root[3]
+                hydra_class = APIClassProcessor.hydra_classes.get(hydra_entity)
+                hydra_collection = APIClassProcessor.hydra_collections.get(hydra_entity)
+
+        return [hydra_class, hydra_collection]

hydra_openapi_parser/hydra_openapi_parser_v2/parsers/resp_parser.py

@@ -0,0 +1,36 @@
+from processors.status_processor import StatusProcessor
+from parsers.schema_parser import SchemaParser
+
+
+class ResponseParser:
+    def __init__(self, code, response) -> None:
+        self.code = code
+        self.response = response
+        self.returns = ""
+
+    def parse_code(self):
+        if self.code.isnumeric():
+            return int(self.code)
+        else:
+            # handles default response
+            return 500
+
+    def parse_returns(self):
+        return self.returns
+
+    def parse(self):
+        response = {"code": self.parse_code(), "desc": "", "title": ""}
+        for key, value in self.response.items():
+            if key == "description":
+                response["desc"] = value
+            if key == "content":
+                for _, expects in value.items():
+                    schema_parser = SchemaParser(expects.get("schema"))
+                    hydra_classes, _ = schema_parser.parse()
+                    for title, _ in hydra_classes.items():
+                        self.returns = title
+
+        status_processor = StatusProcessor(response)
+        hydra_status = status_processor.generate()
+
+        return hydra_status