Refactor parsing

CHANGELOG.md

@@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+
+## 0.5.0 - Unreleased
+### Internal Changes
+- Switched OpenAPI document parsing to use 
+    [openapi-schema-pydantic](https://github.com/kuimono/openapi-schema-pydantic/pull/1) (#103)
+
+
 ## 0.4.2 - 2020-06-13
 ### Additions
 - Support for responses with no content (#63 & #66). Thanks @acgray!

openapi_python_client/__init__.py

@@ -14,7 +14,7 @@
 
 from openapi_python_client import utils
 
-from .openapi_parser import OpenAPI, import_string_from_reference
+from .openapi_parser import GeneratorData, import_string_from_reference
 from .openapi_parser.errors import MultipleParseError
 
 if sys.version_info.minor == 7:  # version did not exist in 3.7, need to use a backport
@@ -28,7 +28,7 @@
 
 def _get_project_for_url_or_path(url: Optional[str], path: Optional[Path]) -> _Project:
     data_dict = _get_json(url=url, path=path)
-    openapi = OpenAPI.from_dict(data_dict)
+    openapi = GeneratorData.from_dict(data_dict)
     return _Project(openapi=openapi)
 
 
@@ -72,8 +72,8 @@ def _get_json(*, url: Optional[str], path: Optional[Path]) -> Dict[str, Any]:
 class _Project:
     TEMPLATE_FILTERS = {"snakecase": utils.snake_case}
 
-    def __init__(self, *, openapi: OpenAPI) -> None:
-        self.openapi: OpenAPI = openapi
+    def __init__(self, *, openapi: GeneratorData) -> None:
+        self.openapi: GeneratorData = openapi
         self.env: Environment = Environment(loader=PackageLoader(__package__), trim_blocks=True, lstrip_blocks=True)
 
         self.project_name: str = f"{openapi.title.replace(' ', '-').lower()}-client"
@@ -170,10 +170,10 @@ def _build_models(self) -> None:
         types_path.write_text(types_template.render())
 
         model_template = self.env.get_template("model.pyi")
-        for schema in self.openapi.schemas.values():
-            module_path = models_dir / f"{schema.reference.module_name}.py"
-            module_path.write_text(model_template.render(schema=schema))
-            imports.append(import_string_from_reference(schema.reference))
+        for model in self.openapi.models.values():
+            module_path = models_dir / f"{model.reference.module_name}.py"
+            module_path.write_text(model_template.render(model=model))
+            imports.append(import_string_from_reference(model.reference))
 
         # Generate enums
         enum_template = self.env.get_template("enum.pyi")

openapi_python_client/openapi_parser/__init__.py

@@ -1,5 +1,5 @@
 """ Classes representing the data in the OpenAPI schema """
 
-__all__ = ["OpenAPI", "import_string_from_reference"]
+__all__ = ["GeneratorData", "import_string_from_reference"]
 
-from .openapi import OpenAPI, import_string_from_reference
+from .openapi import GeneratorData, import_string_from_reference

openapi_python_client/openapi_parser/openapi.py

@@ -2,12 +2,14 @@
 
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List, Optional, Set, Union
+
+import openapi_schema_pydantic as oai
 
 from .errors import ParseError
-from .properties import EnumProperty, Property, property_from_dict
+from .properties import EnumProperty, Property, property_from_data
 from .reference import Reference
-from .responses import ListRefResponse, RefResponse, Response, response_from_dict
+from .responses import ListRefResponse, RefResponse, Response, response_from_data
 
 
 class ParameterLocation(str, Enum):
@@ -32,16 +34,21 @@ class EndpointCollection:
     parse_errors: List[ParseError] = field(default_factory=list)
 
     @staticmethod
-    def from_dict(d: Dict[str, Dict[str, Dict[str, Any]]]) -> Dict[str, EndpointCollection]:
+    def from_data(*, data: Dict[str, oai.PathItem]) -> Dict[str, EndpointCollection]:
         """ Parse the openapi paths data to get EndpointCollections by tag """
         endpoints_by_tag: Dict[str, EndpointCollection] = {}
 
-        for path, path_data in d.items():
-            for method, method_data in path_data.items():
-                tag = method_data.get("tags", ["default"])[0]
+        methods = ["get", "put", "post", "delete", "options", "head", "patch", "trace"]
+
+        for path, path_data in data.items():
+            for method in methods:
+                operation: Optional[oai.Operation] = getattr(path_data, method)
+                if operation is None:
+                    continue
+                tag = (operation.tags or ["default"])[0]
                 collection = endpoints_by_tag.setdefault(tag, EndpointCollection(tag=tag))
                 try:
-                    endpoint = Endpoint.from_data(data=method_data, path=path, method=method, tag=tag)
+                    endpoint = Endpoint.from_data(data=operation, path=path, method=method, tag=tag)
                     collection.endpoints.append(endpoint)
                     collection.relative_imports.update(endpoint.relative_imports)
                 except ParseError as e:
@@ -72,40 +79,40 @@ class Endpoint:
     multipart_body_reference: Optional[Reference] = None
 
     @staticmethod
-    def parse_request_form_body(body: Dict[str, Any]) -> Optional[Reference]:
+    def parse_request_form_body(body: oai.RequestBody) -> Optional[Reference]:
         """ Return form_body_reference """
-        body_content = body["content"]
+        body_content = body.content
         form_body = body_content.get("application/x-www-form-urlencoded")
-        if form_body:
-            return Reference.from_ref(form_body["schema"]["$ref"])
+        if form_body is not None and isinstance(form_body.media_type_schema, oai.Reference):
+            return Reference.from_ref(form_body.media_type_schema.ref)
         return None
 
     @staticmethod
-    def parse_multipart_body(body: Dict[str, Any]) -> Optional[Reference]:
+    def parse_multipart_body(body: oai.RequestBody) -> Optional[Reference]:
         """ Return form_body_reference """
-        body_content = body["content"]
-        body = body_content.get("multipart/form-data")
-        if body:
-            return Reference.from_ref(body["schema"]["$ref"])
+        body_content = body.content
+        json_body = body_content.get("multipart/form-data")
+        if json_body is not None and isinstance(json_body.media_type_schema, oai.Reference):
+            return Reference.from_ref(json_body.media_type_schema.ref)
         return None
 
     @staticmethod
-    def parse_request_json_body(body: Dict[str, Any]) -> Optional[Property]:
+    def parse_request_json_body(body: oai.RequestBody) -> Optional[Property]:
         """ Return json_body """
-        body_content = body["content"]
+        body_content = body.content
         json_body = body_content.get("application/json")
-        if json_body:
-            return property_from_dict("json_body", required=True, data=json_body["schema"])
+        if json_body is not None and json_body.media_type_schema is not None:
+            return property_from_data("json_body", required=True, data=json_body.media_type_schema)
         return None
 
-    def _add_body(self, data: Dict[str, Any]) -> None:
+    def _add_body(self, data: oai.Operation) -> None:
         """ Adds form or JSON body to Endpoint if included in data """
-        if "requestBody" not in data:
+        if data.requestBody is None or isinstance(data.requestBody, oai.Reference):
             return
 
-        self.form_body_reference = Endpoint.parse_request_form_body(data["requestBody"])
-        self.json_body = Endpoint.parse_request_json_body(data["requestBody"])
-        self.multipart_body_reference = Endpoint.parse_multipart_body(data["requestBody"])
+        self.form_body_reference = Endpoint.parse_request_form_body(data.requestBody)
+        self.json_body = Endpoint.parse_request_json_body(data.requestBody)
+        self.multipart_body_reference = Endpoint.parse_multipart_body(data.requestBody)
 
         if self.form_body_reference:
             self.relative_imports.add(import_string_from_reference(self.form_body_reference, prefix="..models"))
@@ -114,50 +121,55 @@ def _add_body(self, data: Dict[str, Any]) -> None:
         if self.json_body is not None:
             self.relative_imports.update(self.json_body.get_imports(prefix="..models"))
 
-    def _add_responses(self, data: Dict[str, Any]) -> None:
-        for code, response_dict in data["responses"].items():
-            response = response_from_dict(status_code=int(code), data=response_dict)
+    def _add_responses(self, data: oai.Responses) -> None:
+        for code, response_data in data.items():
+            response = response_from_data(status_code=int(code), data=response_data)
             if isinstance(response, (RefResponse, ListRefResponse)):
                 self.relative_imports.add(import_string_from_reference(response.reference, prefix="..models"))
             self.responses.append(response)
 
-    def _add_parameters(self, data: Dict[str, Any]) -> None:
-        for param_dict in data.get("parameters", []):
-            prop = property_from_dict(
-                name=param_dict["name"], required=param_dict["required"], data=param_dict["schema"]
-            )
+    def _add_parameters(self, data: oai.Operation) -> None:
+        if data.parameters is None:
+            return
+        for param in data.parameters:
+            if isinstance(param, oai.Reference) or param.param_schema is None:
+                continue
+            prop = property_from_data(name=param.name, required=param.required, data=param.param_schema)
             self.relative_imports.update(prop.get_imports(prefix="..models"))
 
-            if param_dict["in"] == ParameterLocation.QUERY:
+            if param.param_in == ParameterLocation.QUERY:
                 self.query_parameters.append(prop)
-            elif param_dict["in"] == ParameterLocation.PATH:
+            elif param.param_in == ParameterLocation.PATH:
                 self.path_parameters.append(prop)
             else:
-                raise ValueError(f"Don't know where to put this parameter: {param_dict}")
+                raise ValueError(f"Don't know where to put this parameter: {param.dict()}")
 
     @staticmethod
-    def from_data(*, data: Dict[str, Any], path: str, method: str, tag: str) -> Endpoint:
+    def from_data(*, data: oai.Operation, path: str, method: str, tag: str) -> Endpoint:
         """ Construct an endpoint from the OpenAPI data """
 
+        if data.operationId is None:
+            raise ParseError(data=data, message="Path operations with operationId are not yet supported")
+
         endpoint = Endpoint(
             path=path,
             method=method,
-            description=data.get("description"),
-            name=data["operationId"],
-            requires_security=bool(data.get("security")),
+            description=data.description,
+            name=data.operationId,
+            requires_security=bool(data.security),
             tag=tag,
         )
         endpoint._add_parameters(data)
-        endpoint._add_responses(data)
+        endpoint._add_responses(data.responses)
         endpoint._add_body(data)
 
         return endpoint
 
 
 @dataclass
-class Schema:
+class Model:
     """
-    Describes a schema, AKA data model used in requests.
+    A data model used by the API- usually a Schema with type "object".
 
     These will all be converted to dataclasses in the client
     """
@@ -169,72 +181,78 @@ class Schema:
     relative_imports: Set[str]
 
     @staticmethod
-    def from_dict(d: Dict[str, Any], name: str) -> Schema:
-        """ A single Schema from its dict representation
+    def from_data(*, data: Union[oai.Reference, oai.Schema], name: str) -> Model:
+        """ A single Model from its OAI data
 
         Args:
-            d:    Dict representation of the schema
+            data: Data of a single Schema
             name: Name by which the schema is referenced, such as a model name.
                 Used to infer the type name if a `title` property is not available.
         """
-        required_set = set(d.get("required", []))
+        if isinstance(data, oai.Reference):
+            raise ParseError("Reference schemas are not supported.")
+        required_set = set(data.required or [])
         required_properties: List[Property] = []
         optional_properties: List[Property] = []
         relative_imports: Set[str] = set()
 
-        ref = Reference.from_ref(d.get("title", name))
+        ref = Reference.from_ref(data.title or name)
 
-        for key, value in d.get("properties", {}).items():
+        for key, value in (data.properties or {}).items():
             required = key in required_set
-            p = property_from_dict(name=key, required=required, data=value)
+            p = property_from_data(name=key, required=required, data=value)
             if required:
                 required_properties.append(p)
             else:
                 optional_properties.append(p)
             relative_imports.update(p.get_imports(prefix=""))
 
-        schema = Schema(
+        model = Model(
             reference=ref,
             required_properties=required_properties,
             optional_properties=optional_properties,
             relative_imports=relative_imports,
-            description=d.get("description", ""),
+            description=data.description or "",
         )
-        return schema
+        return model
 
     @staticmethod
-    def dict(d: Dict[str, Dict[str, Any]]) -> Dict[str, Schema]:
+    def build(*, schemas: Dict[str, Union[oai.Reference, oai.Schema]]) -> Dict[str, Model]:
         """ Get a list of Schemas from an OpenAPI dict """
         result = {}
-        for name, data in d.items():
-            s = Schema.from_dict(data, name=name)
+        for name, data in schemas.items():
+            s = Model.from_data(data=data, name=name)
             result[s.reference.class_name] = s
         return result
 
 
 @dataclass
-class OpenAPI:
-    """ Top level OpenAPI document """
+class GeneratorData:
+    """ All the data needed to generate a client """
 
     title: str
     description: Optional[str]
     version: str
-    schemas: Dict[str, Schema]
+    models: Dict[str, Model]
     endpoint_collections_by_tag: Dict[str, EndpointCollection]
     enums: Dict[str, EnumProperty]
 
     @staticmethod
-    def from_dict(d: Dict[str, Dict[str, Any]]) -> OpenAPI:
+    def from_dict(d: Dict[str, Dict[str, Any]]) -> GeneratorData:
         """ Create an OpenAPI from dict """
-        schemas = Schema.dict(d["components"]["schemas"])
-        endpoint_collections_by_tag = EndpointCollection.from_dict(d["paths"])
+        openapi = oai.OpenAPI.parse_obj(d)
+        if openapi.components is None or openapi.components.schemas is None:
+            models = {}
+        else:
+            models = Model.build(schemas=openapi.components.schemas)
+        endpoint_collections_by_tag = EndpointCollection.from_data(data=openapi.paths)
         enums = EnumProperty.get_all_enums()
 
-        return OpenAPI(
-            title=d["info"]["title"],
-            description=d["info"].get("description"),
-            version=d["info"]["version"],
+        return GeneratorData(
+            title=openapi.info.title,
+            description=openapi.info.description,
+            version=openapi.info.version,
             endpoint_collections_by_tag=endpoint_collections_by_tag,
-            schemas=schemas,
+            models=models,
             enums=enums,
         )