|
|
|
@@ -47,13 +47,21 @@ class _OASResponseBuilder:
|
|
|
|
|
generate the OAS operation response.
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
def __init__(self, oas_operation):
|
|
|
|
|
def __init__(self, oas_operation, definitions):
|
|
|
|
|
self._oas_operation = oas_operation
|
|
|
|
|
self._definitions = definitions
|
|
|
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
|
def _handle_pydantic_base_model(obj):
|
|
|
|
|
def _process_definitions(self, schema):
|
|
|
|
|
if 'definitions' in schema:
|
|
|
|
|
for k, v in schema['definitions'].items():
|
|
|
|
|
self._definitions[k] = v
|
|
|
|
|
|
|
|
|
|
return {i:schema[i] for i in schema if i!='definitions'}
|
|
|
|
|
|
|
|
|
|
def _handle_pydantic_base_model(self, obj):
|
|
|
|
|
if is_pydantic_base_model(obj):
|
|
|
|
|
return obj.schema()
|
|
|
|
|
return self._process_definitions(obj.schema())
|
|
|
|
|
|
|
|
|
|
return {}
|
|
|
|
|
|
|
|
|
|
def _handle_list(self, obj):
|
|
|
|
@@ -88,7 +96,7 @@ class _OASResponseBuilder:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _add_http_method_to_oas(
|
|
|
|
|
oas_path: PathItem, http_method: str, view: Type[PydanticView]
|
|
|
|
|
oas_path: PathItem, http_method: str, view: Type[PydanticView], definitions: dict
|
|
|
|
|
):
|
|
|
|
|
http_method = http_method.lower()
|
|
|
|
|
oas_operation: OperationObject = getattr(oas_path, http_method)
|
|
|
|
@@ -123,7 +131,7 @@ def _add_http_method_to_oas(
|
|
|
|
|
|
|
|
|
|
return_type = handler.__annotations__.get("return")
|
|
|
|
|
if return_type is not None:
|
|
|
|
|
_OASResponseBuilder(oas_operation).build(return_type)
|
|
|
|
|
_OASResponseBuilder(oas_operation, definitions).build(return_type)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def generate_oas(apps: List[Application]) -> dict:
|
|
|
|
@@ -131,6 +139,7 @@ def generate_oas(apps: List[Application]) -> dict:
|
|
|
|
|
Generate and return Open Api Specification from PydanticView in application.
|
|
|
|
|
"""
|
|
|
|
|
oas = OpenApiSpec3()
|
|
|
|
|
|
|
|
|
|
for app in apps:
|
|
|
|
|
for resources in app.router.resources():
|
|
|
|
|
for resource_route in resources:
|
|
|
|
@@ -140,9 +149,9 @@ def generate_oas(apps: List[Application]) -> dict:
|
|
|
|
|
path = oas.paths[info.get("path", info.get("formatter"))]
|
|
|
|
|
if resource_route.method == "*":
|
|
|
|
|
for method_name in view.allowed_methods:
|
|
|
|
|
_add_http_method_to_oas(path, method_name, view)
|
|
|
|
|
_add_http_method_to_oas(path, method_name, view, oas.definitions)
|
|
|
|
|
else:
|
|
|
|
|
_add_http_method_to_oas(path, resource_route.method, view)
|
|
|
|
|
_add_http_method_to_oas(path, resource_route.method, view, oas.definitions)
|
|
|
|
|
|
|
|
|
|
return oas.spec
|
|
|
|
|
|
|
|
|
@@ -163,6 +172,8 @@ async def oas_ui(request):
|
|
|
|
|
|
|
|
|
|
static_url = request.app.router["static"].url_for(filename="")
|
|
|
|
|
spec_url = request.app.router["spec"].url_for()
|
|
|
|
|
if request.scheme != request.headers.get('x-forwarded-proto', request.scheme):
|
|
|
|
|
request = request.clone(scheme=request.headers['x-forwarded-proto'])
|
|
|
|
|
host = request.url.origin()
|
|
|
|
|
|
|
|
|
|
return Response(
|
|
|
|
|