|
|
@ -7,7 +7,7 @@ import logging
|
|
|
|
import re
|
|
|
|
import re
|
|
|
|
from enum import Enum
|
|
|
|
from enum import Enum
|
|
|
|
from pathlib import Path
|
|
|
|
from pathlib import Path
|
|
|
|
from typing import Dict, List, Optional, Union
|
|
|
|
from typing import TYPE_CHECKING, Dict, List, Optional, Union
|
|
|
|
|
|
|
|
|
|
|
|
import requests
|
|
|
|
import requests
|
|
|
|
import yaml
|
|
|
|
import yaml
|
|
|
@ -39,17 +39,22 @@ class HTTPVerb(str, Enum):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
from openapi_schema_pydantic import (
|
|
|
|
if TYPE_CHECKING:
|
|
|
|
Components,
|
|
|
|
from openapi_schema_pydantic import (
|
|
|
|
OpenAPI,
|
|
|
|
Components,
|
|
|
|
Operation,
|
|
|
|
Operation,
|
|
|
|
Parameter,
|
|
|
|
Parameter,
|
|
|
|
PathItem,
|
|
|
|
PathItem,
|
|
|
|
Paths,
|
|
|
|
Paths,
|
|
|
|
Reference,
|
|
|
|
Reference,
|
|
|
|
RequestBody,
|
|
|
|
RequestBody,
|
|
|
|
Schema,
|
|
|
|
Schema,
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import OpenAPI
|
|
|
|
|
|
|
|
except ImportError:
|
|
|
|
|
|
|
|
OpenAPI = object
|
|
|
|
|
|
|
|
|
|
|
|
class OpenAPISpec(OpenAPI):
|
|
|
|
class OpenAPISpec(OpenAPI):
|
|
|
|
"""OpenAPI Model that removes mis-formatted parts of the spec."""
|
|
|
|
"""OpenAPI Model that removes mis-formatted parts of the spec."""
|
|
|
@ -109,6 +114,8 @@ if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
|
|
|
|
|
|
|
|
def _get_root_referenced_parameter(self, ref: Reference) -> Parameter:
|
|
|
|
def _get_root_referenced_parameter(self, ref: Reference) -> Parameter:
|
|
|
|
"""Get the root reference or err."""
|
|
|
|
"""Get the root reference or err."""
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import Reference
|
|
|
|
|
|
|
|
|
|
|
|
parameter = self._get_referenced_parameter(ref)
|
|
|
|
parameter = self._get_referenced_parameter(ref)
|
|
|
|
while isinstance(parameter, Reference):
|
|
|
|
while isinstance(parameter, Reference):
|
|
|
|
parameter = self._get_referenced_parameter(parameter)
|
|
|
|
parameter = self._get_referenced_parameter(parameter)
|
|
|
@ -123,12 +130,16 @@ if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
return schemas[ref_name]
|
|
|
|
return schemas[ref_name]
|
|
|
|
|
|
|
|
|
|
|
|
def get_schema(self, schema: Union[Reference, Schema]) -> Schema:
|
|
|
|
def get_schema(self, schema: Union[Reference, Schema]) -> Schema:
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import Reference
|
|
|
|
|
|
|
|
|
|
|
|
if isinstance(schema, Reference):
|
|
|
|
if isinstance(schema, Reference):
|
|
|
|
return self.get_referenced_schema(schema)
|
|
|
|
return self.get_referenced_schema(schema)
|
|
|
|
return schema
|
|
|
|
return schema
|
|
|
|
|
|
|
|
|
|
|
|
def _get_root_referenced_schema(self, ref: Reference) -> Schema:
|
|
|
|
def _get_root_referenced_schema(self, ref: Reference) -> Schema:
|
|
|
|
"""Get the root reference or err."""
|
|
|
|
"""Get the root reference or err."""
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import Reference
|
|
|
|
|
|
|
|
|
|
|
|
schema = self.get_referenced_schema(ref)
|
|
|
|
schema = self.get_referenced_schema(ref)
|
|
|
|
while isinstance(schema, Reference):
|
|
|
|
while isinstance(schema, Reference):
|
|
|
|
schema = self.get_referenced_schema(schema)
|
|
|
|
schema = self.get_referenced_schema(schema)
|
|
|
@ -148,6 +159,8 @@ if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
self, ref: Reference
|
|
|
|
self, ref: Reference
|
|
|
|
) -> Optional[RequestBody]:
|
|
|
|
) -> Optional[RequestBody]:
|
|
|
|
"""Get the root request Body or err."""
|
|
|
|
"""Get the root request Body or err."""
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import Reference
|
|
|
|
|
|
|
|
|
|
|
|
request_body = self._get_referenced_request_body(ref)
|
|
|
|
request_body = self._get_referenced_request_body(ref)
|
|
|
|
while isinstance(request_body, Reference):
|
|
|
|
while isinstance(request_body, Reference):
|
|
|
|
request_body = self._get_referenced_request_body(request_body)
|
|
|
|
request_body = self._get_referenced_request_body(request_body)
|
|
|
@ -235,6 +248,8 @@ if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
|
|
|
|
|
|
|
|
def get_methods_for_path(self, path: str) -> List[str]:
|
|
|
|
def get_methods_for_path(self, path: str) -> List[str]:
|
|
|
|
"""Return a list of valid methods for the specified path."""
|
|
|
|
"""Return a list of valid methods for the specified path."""
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import Operation
|
|
|
|
|
|
|
|
|
|
|
|
path_item = self._get_path_strict(path)
|
|
|
|
path_item = self._get_path_strict(path)
|
|
|
|
results = []
|
|
|
|
results = []
|
|
|
|
for method in HTTPVerb:
|
|
|
|
for method in HTTPVerb:
|
|
|
@ -244,6 +259,8 @@ if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
return results
|
|
|
|
return results
|
|
|
|
|
|
|
|
|
|
|
|
def get_parameters_for_path(self, path: str) -> List[Parameter]:
|
|
|
|
def get_parameters_for_path(self, path: str) -> List[Parameter]:
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import Reference
|
|
|
|
|
|
|
|
|
|
|
|
path_item = self._get_path_strict(path)
|
|
|
|
path_item = self._get_path_strict(path)
|
|
|
|
parameters = []
|
|
|
|
parameters = []
|
|
|
|
if not path_item.parameters:
|
|
|
|
if not path_item.parameters:
|
|
|
@ -256,6 +273,8 @@ if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
|
|
|
|
|
|
|
|
def get_operation(self, path: str, method: str) -> Operation:
|
|
|
|
def get_operation(self, path: str, method: str) -> Operation:
|
|
|
|
"""Get the operation object for a given path and HTTP method."""
|
|
|
|
"""Get the operation object for a given path and HTTP method."""
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import Operation
|
|
|
|
|
|
|
|
|
|
|
|
path_item = self._get_path_strict(path)
|
|
|
|
path_item = self._get_path_strict(path)
|
|
|
|
operation_obj = getattr(path_item, method, None)
|
|
|
|
operation_obj = getattr(path_item, method, None)
|
|
|
|
if not isinstance(operation_obj, Operation):
|
|
|
|
if not isinstance(operation_obj, Operation):
|
|
|
@ -264,6 +283,8 @@ if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
|
|
|
|
|
|
|
|
def get_parameters_for_operation(self, operation: Operation) -> List[Parameter]:
|
|
|
|
def get_parameters_for_operation(self, operation: Operation) -> List[Parameter]:
|
|
|
|
"""Get the components for a given operation."""
|
|
|
|
"""Get the components for a given operation."""
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import Reference
|
|
|
|
|
|
|
|
|
|
|
|
parameters = []
|
|
|
|
parameters = []
|
|
|
|
if operation.parameters:
|
|
|
|
if operation.parameters:
|
|
|
|
for parameter in operation.parameters:
|
|
|
|
for parameter in operation.parameters:
|
|
|
@ -276,6 +297,8 @@ if _PYDANTIC_MAJOR_VERSION == 1:
|
|
|
|
self, operation: Operation
|
|
|
|
self, operation: Operation
|
|
|
|
) -> Optional[RequestBody]:
|
|
|
|
) -> Optional[RequestBody]:
|
|
|
|
"""Get the request body for a given operation."""
|
|
|
|
"""Get the request body for a given operation."""
|
|
|
|
|
|
|
|
from openapi_schema_pydantic import Reference
|
|
|
|
|
|
|
|
|
|
|
|
request_body = operation.requestBody
|
|
|
|
request_body = operation.requestBody
|
|
|
|
if isinstance(request_body, Reference):
|
|
|
|
if isinstance(request_body, Reference):
|
|
|
|
request_body = self._get_root_referenced_request_body(request_body)
|
|
|
|
request_body = self._get_root_referenced_request_body(request_body)
|
|
|
|