pub/aiohttp-pydantic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into fixed

Browse Source 
# Conflicts:
#	aiohttp_pydantic/oas/__init__.py
#	aiohttp_pydantic/view.py
This commit is contained in:
Georg K
2022-03-30 17:30:15 +03:00
parent 16ba8caa5b 69fb553635
commit 27c0d76e16
36 changed files with 2068 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
aiohttp_pydantic/oas/view.py
View File

@@ -1,13 +1,14 @@
import typing
from inspect import getdoc
from itertools import count
from typing import List, Type
from typing import List, Type, Optional, get_type_hints
from aiohttp.web import Response, json_response
from aiohttp.web_app import Application
from pydantic import BaseModel
from aiohttp_pydantic.oas.struct import OpenApiSpec3, OperationObject, PathItem
from . import docstring_parser
from ..injectors import _parse_func_signature
from ..utils import is_pydantic_base_model
@@ -15,35 +16,23 @@ from ..view import PydanticView, is_pydantic_view
from .typing import is_status_code_type
def _handle_optional(type_):
"""
Returns the type wrapped in Optional or None.
>>> _handle_optional(int)
>>> _handle_optional(Optional[str])
<class 'str'>
"""
if typing.get_origin(type_) is typing.Union:
args = typing.get_args(type_)
if len(args) == 2 and type(None) in args:
return next(iter(set(args) - {type(None)}))
return None
class _OASResponseBuilder:
"""
Parse the type annotated as returned by a function and
generate the OAS operation response.
"""
def __init__(self, oas: OpenApiSpec3, oas_operation):
def __init__(self, oas: OpenApiSpec3, oas_operation, status_code_descriptions):
self._oas_operation = oas_operation
self._oas = oas
self._status_code_descriptions = status_code_descriptions
def _handle_pydantic_base_model(self, obj):
if is_pydantic_base_model(obj):
response_schema = obj.schema(ref_template="#/components/schemas/{model}")
if def_sub_schemas := response_schema.get("definitions", None):
response_schema = obj.schema(
ref_template="#/components/schemas/{model}"
).copy()
if def_sub_schemas := response_schema.pop("definitions", None):
self._oas.components.schemas.update(def_sub_schemas)
return response_schema
return {}
@@ -64,10 +53,16 @@ class _OASResponseBuilder:
"schema": self._handle_list(typing.get_args(obj)[0])
}
}
desc = self._status_code_descriptions.get(int(status_code))
if desc:
self._oas_operation.responses[status_code].description = desc
elif is_status_code_type(obj):
status_code = obj.__name__[1:]
self._oas_operation.responses[status_code].content = {}
desc = self._status_code_descriptions.get(int(status_code))
if desc:
self._oas_operation.responses[status_code].description = desc
def _handle_union(self, obj):
if typing.get_origin(obj) is typing.Union:
@@ -86,17 +81,23 @@ def _add_http_method_to_oas(
oas_operation: OperationObject = getattr(oas_path, http_method)
handler = getattr(view, http_method)
path_args, body_args, qs_args, header_args, defaults = _parse_func_signature(
handler
handler, unpack_group=True
)
description = getdoc(handler)
if description:
oas_operation.description = description
oas_operation.description = docstring_parser.operation(description)
oas_operation.tags = docstring_parser.tags(description)
status_code_descriptions = docstring_parser.status_code(description)
else:
status_code_descriptions = {}
if body_args:
body_schema = next(iter(body_args.values())).schema(
ref_template="#/components/schemas/{model}"
body_schema = (
next(iter(body_args.values()))
.schema(ref_template="#/components/schemas/{model}")
.copy()
)
if def_sub_schemas := body_schema.get("definitions", None):
if def_sub_schemas := body_schema.pop("definitions", None):
oas.components.schemas.update(def_sub_schemas)
oas_operation.request_body.content = {
@@ -113,28 +114,41 @@ def _add_http_method_to_oas(
i = next(indexes)
oas_operation.parameters[i].in_ = args_location
oas_operation.parameters[i].name = name
optional_type = _handle_optional(type_)
attrs = {"__annotations__": {"__root__": type_}}
if name in defaults:
attrs["__root__"] = defaults[name]
oas_operation.parameters[i].required = False
else:
oas_operation.parameters[i].required = True
oas_operation.parameters[i].schema = type(name, (BaseModel,), attrs).schema(
ref_template="#/components/schemas/{model}"
)
oas_operation.parameters[i].required = optional_type is None
return_type = handler.__annotations__.get("return")
return_type = get_type_hints(handler).get("return")
if return_type is not None:
_OASResponseBuilder(oas, oas_operation).build(return_type)
_OASResponseBuilder(oas, oas_operation, status_code_descriptions).build(
return_type
)
def generate_oas(apps: List[Application]) -> dict:
def generate_oas(
apps: List[Application],
version_spec: Optional[str] = None,
title_spec: Optional[str] = None,
) -> dict:
"""
Generate and return Open Api Specification from PydanticView in application.
"""
oas = OpenApiSpec3()
if version_spec is not None:
oas.info.version = version_spec
if title_spec is not None:
oas.info.title = title_spec
for app in apps:
for resources in app.router.resources():
for resource_route in resources:
@@ -158,7 +172,9 @@ async def get_oas(request):
View to generate the Open Api Specification from PydanticView in application.
"""
apps = request.app["apps to expose"]
return json_response(generate_oas(apps))
version_spec = request.app["version_spec"]
title_spec = request.app["title_spec"]
return json_response(generate_oas(apps, version_spec, title_spec))
async def oas_ui(request):