Compare commits
1 Commits
Author | SHA1 | Date | |
---|---|---|---|
|
2a064a75d9 |
80
README.rst
80
README.rst
@@ -1,18 +1,6 @@
|
|||||||
Aiohttp pydantic - Aiohttp View to validate and parse request
|
Aiohttp pydantic - Aiohttp View to validate and parse request
|
||||||
=============================================================
|
=============================================================
|
||||||
|
|
||||||
|
|
||||||
Aiohttp pydantic is an `aiohttp view`_ to easily parse and validate request.
|
|
||||||
You define using the function annotations what your methods for handling HTTP verbs expects and Aiohttp pydantic parses the HTTP request
|
|
||||||
for you, validates the data, and injects that you want as parameters.
|
|
||||||
|
|
||||||
|
|
||||||
Features:
|
|
||||||
|
|
||||||
- Query string, request body, URL path and HTTP headers validation.
|
|
||||||
- Open API Specification generation.
|
|
||||||
|
|
||||||
|
|
||||||
How to install
|
How to install
|
||||||
--------------
|
--------------
|
||||||
|
|
||||||
@@ -157,8 +145,8 @@ To declare a HTTP headers parameters, you must declare your argument as a `keywo
|
|||||||
.. _pydantic Model: https://pydantic-docs.helpmanual.io/usage/models/
|
.. _pydantic Model: https://pydantic-docs.helpmanual.io/usage/models/
|
||||||
.. _keyword-only argument: https://www.python.org/dev/peps/pep-3102/
|
.. _keyword-only argument: https://www.python.org/dev/peps/pep-3102/
|
||||||
|
|
||||||
Add route to generate Open Api Specification (OAS)
|
Add route to generate Open Api Specification
|
||||||
--------------------------------------------------
|
--------------------------------------------
|
||||||
|
|
||||||
aiohttp_pydantic provides a sub-application to serve a route to generate Open Api Specification
|
aiohttp_pydantic provides a sub-application to serve a route to generate Open Api Specification
|
||||||
reading annotation in your PydanticView. Use *aiohttp_pydantic.oas.setup()* to add the sub-application
|
reading annotation in your PydanticView. Use *aiohttp_pydantic.oas.setup()* to add the sub-application
|
||||||
@@ -180,8 +168,8 @@ By default, the route to display the Open Api Specification is /oas but you can
|
|||||||
|
|
||||||
oas.setup(app, url_prefix='/spec-api')
|
oas.setup(app, url_prefix='/spec-api')
|
||||||
|
|
||||||
If you want generate the Open Api Specification from specific aiohttp sub-applications.
|
If you want generate the Open Api Specification from several aiohttp sub-application.
|
||||||
on the same route, you must use *apps_to_expose* parameter.
|
on the same route, you must use *apps_to_expose* parameters
|
||||||
|
|
||||||
|
|
||||||
.. code-block:: python3
|
.. code-block:: python3
|
||||||
@@ -191,61 +179,11 @@ on the same route, you must use *apps_to_expose* parameter.
|
|||||||
|
|
||||||
app = web.Application()
|
app = web.Application()
|
||||||
sub_app_1 = web.Application()
|
sub_app_1 = web.Application()
|
||||||
sub_app_2 = web.Application()
|
|
||||||
|
|
||||||
oas.setup(app, apps_to_expose=[sub_app_1, sub_app_2])
|
oas.setup(app, apps_to_expose=[app, sub_app_1])
|
||||||
|
|
||||||
Add annotation to define response content
|
|
||||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
||||||
|
|
||||||
The module aiohttp_pydantic.oas.typing provides class to annotate a
|
|
||||||
response content.
|
|
||||||
|
|
||||||
For example *r200[List[Pet]]* means the server responses with
|
|
||||||
the status code 200 and the response content is a List of Pet where Pet will be
|
|
||||||
defined using a pydantic.BaseModel
|
|
||||||
|
|
||||||
|
|
||||||
.. code-block:: python3
|
|
||||||
|
|
||||||
from aiohttp_pydantic import PydanticView
|
|
||||||
from aiohttp_pydantic.oas.typing import r200, r201, r204, r404
|
|
||||||
|
|
||||||
|
|
||||||
class Pet(BaseModel):
|
|
||||||
id: int
|
|
||||||
name: str
|
|
||||||
|
|
||||||
|
|
||||||
class Error(BaseModel):
|
|
||||||
error: str
|
|
||||||
|
|
||||||
|
|
||||||
class PetCollectionView(PydanticView):
|
|
||||||
async def get(self) -> r200[List[Pet]]:
|
|
||||||
pets = self.request.app["model"].list_pets()
|
|
||||||
return web.json_response([pet.dict() for pet in pets])
|
|
||||||
|
|
||||||
async def post(self, pet: Pet) -> r201[Pet]:
|
|
||||||
self.request.app["model"].add_pet(pet)
|
|
||||||
return web.json_response(pet.dict())
|
|
||||||
|
|
||||||
|
|
||||||
class PetItemView(PydanticView):
|
|
||||||
async def get(self, id: int, /) -> Union[r200[Pet], r404[Error]]:
|
|
||||||
pet = self.request.app["model"].find_pet(id)
|
|
||||||
return web.json_response(pet.dict())
|
|
||||||
|
|
||||||
async def put(self, id: int, /, pet: Pet) -> r200[Pet]:
|
|
||||||
self.request.app["model"].update_pet(id, pet)
|
|
||||||
return web.json_response(pet.dict())
|
|
||||||
|
|
||||||
async def delete(self, id: int, /) -> r204:
|
|
||||||
self.request.app["model"].remove_pet(id)
|
|
||||||
return web.Response(status=204)
|
|
||||||
|
|
||||||
Demo
|
Demo
|
||||||
----
|
====
|
||||||
|
|
||||||
Have a look at `demo`_ for a complete example
|
Have a look at `demo`_ for a complete example
|
||||||
|
|
||||||
@@ -258,12 +196,6 @@ Have a look at `demo`_ for a complete example
|
|||||||
|
|
||||||
Go to http://127.0.0.1:8080/oas
|
Go to http://127.0.0.1:8080/oas
|
||||||
|
|
||||||
You can generate the OAS in a json file using the command:
|
|
||||||
|
|
||||||
.. code-block:: bash
|
|
||||||
|
|
||||||
python -m aiohttp_pydantic.oas demo.main
|
|
||||||
|
|
||||||
|
|
||||||
.. _demo: https://github.com/Maillol/aiohttp-pydantic/tree/main/demo
|
.. _demo: https://github.com/Maillol/aiohttp-pydantic/tree/main/demo
|
||||||
.. _aiohttp view: https://docs.aiohttp.org/en/stable/web_quickstart.html#class-based-views
|
|
||||||
|
@@ -1,5 +1,5 @@
|
|||||||
from .view import PydanticView
|
from .view import PydanticView
|
||||||
|
|
||||||
__version__ = "1.4.1"
|
__version__ = "1.1.0"
|
||||||
|
|
||||||
__all__ = ("PydanticView", "__version__")
|
__all__ = ("PydanticView", "__version__")
|
||||||
|
@@ -1,13 +1,11 @@
|
|||||||
import abc
|
|
||||||
from inspect import signature
|
|
||||||
from json.decoder import JSONDecodeError
|
|
||||||
from typing import Callable, Tuple
|
from typing import Callable, Tuple
|
||||||
|
|
||||||
from aiohttp.web_exceptions import HTTPBadRequest
|
|
||||||
from aiohttp.web_request import BaseRequest
|
from aiohttp.web_request import BaseRequest
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
|
from inspect import signature
|
||||||
|
|
||||||
from .utils import is_pydantic_base_model
|
|
||||||
|
import abc
|
||||||
|
|
||||||
|
|
||||||
class AbstractInjector(metaclass=abc.ABCMeta):
|
class AbstractInjector(metaclass=abc.ABCMeta):
|
||||||
@@ -49,13 +47,7 @@ class BodyGetter(AbstractInjector):
|
|||||||
self.arg_name, self.model = next(iter(args_spec.items()))
|
self.arg_name, self.model = next(iter(args_spec.items()))
|
||||||
|
|
||||||
async def inject(self, request: BaseRequest, args_view: list, kwargs_view: dict):
|
async def inject(self, request: BaseRequest, args_view: list, kwargs_view: dict):
|
||||||
try:
|
body = await request.json()
|
||||||
body = await request.json()
|
|
||||||
except JSONDecodeError:
|
|
||||||
raise HTTPBadRequest(
|
|
||||||
text='{"error": "Malformed JSON"}', content_type="application/json"
|
|
||||||
)
|
|
||||||
|
|
||||||
kwargs_view[self.arg_name] = self.model(**body)
|
kwargs_view[self.arg_name] = self.model(**body)
|
||||||
|
|
||||||
|
|
||||||
@@ -108,7 +100,7 @@ def _parse_func_signature(func: Callable) -> Tuple[dict, dict, dict, dict]:
|
|||||||
if param_spec.kind is param_spec.POSITIONAL_ONLY:
|
if param_spec.kind is param_spec.POSITIONAL_ONLY:
|
||||||
path_args[param_name] = param_spec.annotation
|
path_args[param_name] = param_spec.annotation
|
||||||
elif param_spec.kind is param_spec.POSITIONAL_OR_KEYWORD:
|
elif param_spec.kind is param_spec.POSITIONAL_OR_KEYWORD:
|
||||||
if is_pydantic_base_model(param_spec.annotation):
|
if issubclass(param_spec.annotation, BaseModel):
|
||||||
body_args[param_name] = param_spec.annotation
|
body_args[param_name] = param_spec.annotation
|
||||||
else:
|
else:
|
||||||
qs_args[param_name] = param_spec.annotation
|
qs_args[param_name] = param_spec.annotation
|
||||||
|
@@ -1,11 +1,10 @@
|
|||||||
from importlib import resources
|
|
||||||
from typing import Iterable
|
from typing import Iterable
|
||||||
|
from importlib import resources
|
||||||
|
|
||||||
import jinja2
|
import jinja2
|
||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
from swagger_ui_bundle import swagger_ui_path
|
|
||||||
|
|
||||||
from .view import get_oas, oas_ui
|
from .view import get_oas, oas_ui
|
||||||
|
from swagger_ui_bundle import swagger_ui_path
|
||||||
|
|
||||||
|
|
||||||
def setup(
|
def setup(
|
||||||
|
@@ -1,8 +0,0 @@
|
|||||||
import argparse
|
|
||||||
|
|
||||||
from .cmd import setup
|
|
||||||
|
|
||||||
parser = argparse.ArgumentParser(description="Generate Open API Specification")
|
|
||||||
setup(parser)
|
|
||||||
args = parser.parse_args()
|
|
||||||
args.func(args)
|
|
@@ -1,45 +0,0 @@
|
|||||||
import argparse
|
|
||||||
import importlib
|
|
||||||
import json
|
|
||||||
|
|
||||||
from .view import generate_oas
|
|
||||||
|
|
||||||
|
|
||||||
def application_type(value):
|
|
||||||
"""
|
|
||||||
Return aiohttp application defined in the value.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
module_name, app_name = value.split(":")
|
|
||||||
except ValueError:
|
|
||||||
module_name, app_name = value, "app"
|
|
||||||
|
|
||||||
module = importlib.import_module(module_name)
|
|
||||||
try:
|
|
||||||
if app_name.endswith("()"):
|
|
||||||
app_name = app_name.strip("()")
|
|
||||||
factory_app = getattr(module, app_name)
|
|
||||||
return factory_app()
|
|
||||||
return getattr(module, app_name)
|
|
||||||
|
|
||||||
except AttributeError as error:
|
|
||||||
raise argparse.ArgumentTypeError(error) from error
|
|
||||||
|
|
||||||
|
|
||||||
def setup(parser: argparse.ArgumentParser):
|
|
||||||
parser.add_argument(
|
|
||||||
"apps",
|
|
||||||
metavar="APP",
|
|
||||||
type=application_type,
|
|
||||||
nargs="*",
|
|
||||||
help="The name of the module containing the asyncio.web.Application."
|
|
||||||
" By default the variable named 'app' is loaded but you can define"
|
|
||||||
" an other variable name ending the name of module with : characters"
|
|
||||||
" and the name of variable. Example: my_package.my_module:my_app",
|
|
||||||
)
|
|
||||||
|
|
||||||
parser.set_defaults(func=show_oas)
|
|
||||||
|
|
||||||
|
|
||||||
def show_oas(args: argparse.Namespace):
|
|
||||||
print(json.dumps(generate_oas(args.apps), sort_keys=True, indent=4))
|
|
@@ -1,6 +1,3 @@
|
|||||||
from typing import Union
|
|
||||||
|
|
||||||
|
|
||||||
class Info:
|
class Info:
|
||||||
def __init__(self, spec: dict):
|
def __init__(self, spec: dict):
|
||||||
self._spec = spec.setdefault("info", {})
|
self._spec = spec.setdefault("info", {})
|
||||||
@@ -118,39 +115,6 @@ class Parameters:
|
|||||||
return Parameter(spec)
|
return Parameter(spec)
|
||||||
|
|
||||||
|
|
||||||
class Response:
|
|
||||||
def __init__(self, spec: dict):
|
|
||||||
self._spec = spec
|
|
||||||
|
|
||||||
@property
|
|
||||||
def description(self) -> str:
|
|
||||||
return self._spec["description"]
|
|
||||||
|
|
||||||
@description.setter
|
|
||||||
def description(self, description: str):
|
|
||||||
self._spec["description"] = description
|
|
||||||
|
|
||||||
@property
|
|
||||||
def content(self):
|
|
||||||
return self._spec["content"]
|
|
||||||
|
|
||||||
@content.setter
|
|
||||||
def content(self, content: dict):
|
|
||||||
self._spec["content"] = content
|
|
||||||
|
|
||||||
|
|
||||||
class Responses:
|
|
||||||
def __init__(self, spec: dict):
|
|
||||||
self._spec = spec.setdefault("responses", {})
|
|
||||||
|
|
||||||
def __getitem__(self, status_code: Union[int, str]) -> Response:
|
|
||||||
if not (100 <= int(status_code) < 600):
|
|
||||||
raise ValueError("status_code must be between 100 and 599")
|
|
||||||
|
|
||||||
spec = self._spec.setdefault(str(status_code), {})
|
|
||||||
return Response(spec)
|
|
||||||
|
|
||||||
|
|
||||||
class OperationObject:
|
class OperationObject:
|
||||||
def __init__(self, spec: dict):
|
def __init__(self, spec: dict):
|
||||||
self._spec = spec
|
self._spec = spec
|
||||||
@@ -179,10 +143,6 @@ class OperationObject:
|
|||||||
def parameters(self) -> Parameters:
|
def parameters(self) -> Parameters:
|
||||||
return Parameters(self._spec)
|
return Parameters(self._spec)
|
||||||
|
|
||||||
@property
|
|
||||||
def responses(self) -> Responses:
|
|
||||||
return Responses(self._spec)
|
|
||||||
|
|
||||||
|
|
||||||
class PathItem:
|
class PathItem:
|
||||||
def __init__(self, spec: dict):
|
def __init__(self, spec: dict):
|
||||||
|
@@ -1,47 +0,0 @@
|
|||||||
"""
|
|
||||||
This module provides type to annotate the content of web.Response returned by
|
|
||||||
the HTTP handlers.
|
|
||||||
|
|
||||||
The type are: r100, r101, ..., r599
|
|
||||||
|
|
||||||
Example:
|
|
||||||
|
|
||||||
class PetCollectionView(PydanticView):
|
|
||||||
async def get(self) -> Union[r200[List[Pet]], r404]:
|
|
||||||
...
|
|
||||||
"""
|
|
||||||
|
|
||||||
from functools import lru_cache
|
|
||||||
from types import new_class
|
|
||||||
from typing import Protocol, TypeVar
|
|
||||||
|
|
||||||
RespContents = TypeVar("RespContents", covariant=True)
|
|
||||||
|
|
||||||
_status_code = frozenset(f"r{code}" for code in range(100, 600))
|
|
||||||
|
|
||||||
|
|
||||||
@lru_cache(maxsize=len(_status_code))
|
|
||||||
def _make_status_code_type(status_code):
|
|
||||||
if status_code in _status_code:
|
|
||||||
return new_class(status_code, (Protocol[RespContents],))
|
|
||||||
|
|
||||||
|
|
||||||
def is_status_code_type(obj):
|
|
||||||
"""
|
|
||||||
Return True if obj is a status code type such as _200 or _404.
|
|
||||||
"""
|
|
||||||
name = getattr(obj, "__name__", None)
|
|
||||||
if name not in _status_code:
|
|
||||||
return False
|
|
||||||
|
|
||||||
return obj is _make_status_code_type(name)
|
|
||||||
|
|
||||||
|
|
||||||
def __getattr__(name):
|
|
||||||
if (status_code_type := _make_status_code_type(name)) is None:
|
|
||||||
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
|
||||||
return status_code_type
|
|
||||||
|
|
||||||
|
|
||||||
__all__ = list(_status_code)
|
|
||||||
__all__.append("is_status_code_type")
|
|
@@ -1,106 +1,51 @@
|
|||||||
import typing
|
from aiohttp.web import json_response, Response
|
||||||
from typing import List, Type
|
|
||||||
|
|
||||||
from aiohttp.web import Response, json_response
|
|
||||||
from aiohttp.web_app import Application
|
|
||||||
|
|
||||||
from aiohttp_pydantic.oas.struct import OpenApiSpec3, OperationObject, PathItem
|
from aiohttp_pydantic.oas.struct import OpenApiSpec3, OperationObject, PathItem
|
||||||
|
from typing import Type
|
||||||
|
|
||||||
from ..injectors import _parse_func_signature
|
from ..injectors import _parse_func_signature
|
||||||
from ..utils import is_pydantic_base_model
|
|
||||||
from ..view import PydanticView, is_pydantic_view
|
from ..view import PydanticView, is_pydantic_view
|
||||||
from .typing import is_status_code_type
|
|
||||||
|
|
||||||
JSON_SCHEMA_TYPES = {float: "number", str: "string", int: "integer"}
|
JSON_SCHEMA_TYPES = {float: "number", str: "string", int: "integer"}
|
||||||
|
|
||||||
|
|
||||||
class _OASResponseBuilder:
|
def _add_http_method_to_oas(oas_path: PathItem, method: str, view: Type[PydanticView]):
|
||||||
"""
|
method = method.lower()
|
||||||
Parse the type annotated as returned by a function and
|
mtd: OperationObject = getattr(oas_path, method)
|
||||||
generate the OAS operation response.
|
handler = getattr(view, method)
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, oas_operation):
|
|
||||||
self._oas_operation = oas_operation
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _handle_pydantic_base_model(obj):
|
|
||||||
if is_pydantic_base_model(obj):
|
|
||||||
return obj.schema()
|
|
||||||
return {}
|
|
||||||
|
|
||||||
def _handle_list(self, obj):
|
|
||||||
if typing.get_origin(obj) is list:
|
|
||||||
return {
|
|
||||||
"type": "array",
|
|
||||||
"items": self._handle_pydantic_base_model(typing.get_args(obj)[0]),
|
|
||||||
}
|
|
||||||
return self._handle_pydantic_base_model(obj)
|
|
||||||
|
|
||||||
def _handle_status_code_type(self, obj):
|
|
||||||
if is_status_code_type(typing.get_origin(obj)):
|
|
||||||
status_code = typing.get_origin(obj).__name__[1:]
|
|
||||||
self._oas_operation.responses[status_code].content = {
|
|
||||||
"application/json": {
|
|
||||||
"schema": self._handle_list(typing.get_args(obj)[0])
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
elif is_status_code_type(obj):
|
|
||||||
status_code = obj.__name__[1:]
|
|
||||||
self._oas_operation.responses[status_code].content = {}
|
|
||||||
|
|
||||||
def _handle_union(self, obj):
|
|
||||||
if typing.get_origin(obj) is typing.Union:
|
|
||||||
for arg in typing.get_args(obj):
|
|
||||||
self._handle_status_code_type(arg)
|
|
||||||
self._handle_status_code_type(obj)
|
|
||||||
|
|
||||||
def build(self, obj):
|
|
||||||
self._handle_union(obj)
|
|
||||||
|
|
||||||
|
|
||||||
def _add_http_method_to_oas(
|
|
||||||
oas_path: PathItem, http_method: str, view: Type[PydanticView]
|
|
||||||
):
|
|
||||||
http_method = http_method.lower()
|
|
||||||
oas_operation: OperationObject = getattr(oas_path, http_method)
|
|
||||||
handler = getattr(view, http_method)
|
|
||||||
path_args, body_args, qs_args, header_args = _parse_func_signature(handler)
|
path_args, body_args, qs_args, header_args = _parse_func_signature(handler)
|
||||||
|
|
||||||
if body_args:
|
if body_args:
|
||||||
oas_operation.request_body.content = {
|
mtd.request_body.content = {
|
||||||
"application/json": {"schema": next(iter(body_args.values())).schema()}
|
"application/json": {"schema": next(iter(body_args.values())).schema()}
|
||||||
}
|
}
|
||||||
|
|
||||||
i = 0
|
i = 0
|
||||||
for i, (name, type_) in enumerate(path_args.items()):
|
for i, (name, type_) in enumerate(path_args.items()):
|
||||||
oas_operation.parameters[i].required = True
|
mtd.parameters[i].required = True
|
||||||
oas_operation.parameters[i].in_ = "path"
|
mtd.parameters[i].in_ = "path"
|
||||||
oas_operation.parameters[i].name = name
|
mtd.parameters[i].name = name
|
||||||
oas_operation.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
||||||
|
|
||||||
for i, (name, type_) in enumerate(qs_args.items(), i + 1):
|
for i, (name, type_) in enumerate(qs_args.items(), i + 1):
|
||||||
oas_operation.parameters[i].required = False
|
mtd.parameters[i].required = False
|
||||||
oas_operation.parameters[i].in_ = "query"
|
mtd.parameters[i].in_ = "query"
|
||||||
oas_operation.parameters[i].name = name
|
mtd.parameters[i].name = name
|
||||||
oas_operation.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
||||||
|
|
||||||
for i, (name, type_) in enumerate(header_args.items(), i + 1):
|
for i, (name, type_) in enumerate(header_args.items(), i + 1):
|
||||||
oas_operation.parameters[i].required = False
|
mtd.parameters[i].required = False
|
||||||
oas_operation.parameters[i].in_ = "header"
|
mtd.parameters[i].in_ = "header"
|
||||||
oas_operation.parameters[i].name = name
|
mtd.parameters[i].name = name
|
||||||
oas_operation.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
||||||
|
|
||||||
return_type = handler.__annotations__.get("return")
|
|
||||||
if return_type is not None:
|
|
||||||
_OASResponseBuilder(oas_operation).build(return_type)
|
|
||||||
|
|
||||||
|
|
||||||
def generate_oas(apps: List[Application]) -> dict:
|
async def get_oas(request):
|
||||||
"""
|
"""
|
||||||
Generate and return Open Api Specification from PydanticView in application.
|
Generate Open Api Specification from PydanticView in application.
|
||||||
"""
|
"""
|
||||||
|
apps = request.app["apps to expose"]
|
||||||
oas = OpenApiSpec3()
|
oas = OpenApiSpec3()
|
||||||
for app in apps:
|
for app in apps:
|
||||||
for resources in app.router.resources():
|
for resources in app.router.resources():
|
||||||
@@ -115,15 +60,7 @@ def generate_oas(apps: List[Application]) -> dict:
|
|||||||
else:
|
else:
|
||||||
_add_http_method_to_oas(path, resource_route.method, view)
|
_add_http_method_to_oas(path, resource_route.method, view)
|
||||||
|
|
||||||
return oas.spec
|
return json_response(oas.spec)
|
||||||
|
|
||||||
|
|
||||||
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))
|
|
||||||
|
|
||||||
|
|
||||||
async def oas_ui(request):
|
async def oas_ui(request):
|
||||||
|
@@ -1,11 +0,0 @@
|
|||||||
from pydantic import BaseModel
|
|
||||||
|
|
||||||
|
|
||||||
def is_pydantic_base_model(obj):
|
|
||||||
"""
|
|
||||||
Return true is obj is a pydantic.BaseModel subclass.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
return issubclass(obj, BaseModel)
|
|
||||||
except TypeError:
|
|
||||||
return False
|
|
@@ -1,16 +1,22 @@
|
|||||||
from functools import update_wrapper
|
|
||||||
from inspect import iscoroutinefunction
|
from inspect import iscoroutinefunction
|
||||||
from typing import Any, Callable, Generator, Iterable
|
|
||||||
|
|
||||||
from aiohttp.abc import AbstractView
|
from aiohttp.abc import AbstractView
|
||||||
from aiohttp.hdrs import METH_ALL
|
from aiohttp.hdrs import METH_ALL
|
||||||
from aiohttp.web import json_response
|
|
||||||
from aiohttp.web_exceptions import HTTPMethodNotAllowed
|
from aiohttp.web_exceptions import HTTPMethodNotAllowed
|
||||||
from aiohttp.web_response import StreamResponse
|
from aiohttp.web_response import StreamResponse
|
||||||
from pydantic import ValidationError
|
from pydantic import ValidationError
|
||||||
|
from typing import Generator, Any, Callable, Type, Iterable
|
||||||
|
from aiohttp.web import json_response
|
||||||
|
from functools import update_wrapper
|
||||||
|
|
||||||
from .injectors import (AbstractInjector, BodyGetter, HeadersGetter,
|
|
||||||
MatchInfoGetter, QueryGetter, _parse_func_signature)
|
from .injectors import (
|
||||||
|
MatchInfoGetter,
|
||||||
|
HeadersGetter,
|
||||||
|
QueryGetter,
|
||||||
|
BodyGetter,
|
||||||
|
AbstractInjector,
|
||||||
|
_parse_func_signature,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class PydanticView(AbstractView):
|
class PydanticView(AbstractView):
|
||||||
|
@@ -1,5 +1,25 @@
|
|||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
|
|
||||||
from .main import app
|
from aiohttp_pydantic import oas
|
||||||
|
from aiohttp.web import middleware
|
||||||
|
|
||||||
|
from .view import PetItemView, PetCollectionView
|
||||||
|
from .model import Model
|
||||||
|
|
||||||
|
|
||||||
|
@middleware
|
||||||
|
async def pet_not_found_to_404(request, handler):
|
||||||
|
try:
|
||||||
|
return await handler(request)
|
||||||
|
except Model.NotFound as key:
|
||||||
|
return web.json_response({"error": f"Pet {key} does not exist"}, status=404)
|
||||||
|
|
||||||
|
|
||||||
|
app = web.Application(middlewares=[pet_not_found_to_404])
|
||||||
|
oas.setup(app)
|
||||||
|
|
||||||
|
app["model"] = Model()
|
||||||
|
app.router.add_view("/pets", PetCollectionView)
|
||||||
|
app.router.add_view("/pets/{id}", PetItemView)
|
||||||
|
|
||||||
web.run_app(app)
|
web.run_app(app)
|
||||||
|
22
demo/main.py
22
demo/main.py
@@ -1,22 +0,0 @@
|
|||||||
from aiohttp.web import Application, json_response, middleware
|
|
||||||
|
|
||||||
from aiohttp_pydantic import oas
|
|
||||||
|
|
||||||
from .model import Model
|
|
||||||
from .view import PetCollectionView, PetItemView
|
|
||||||
|
|
||||||
|
|
||||||
@middleware
|
|
||||||
async def pet_not_found_to_404(request, handler):
|
|
||||||
try:
|
|
||||||
return await handler(request)
|
|
||||||
except Model.NotFound as key:
|
|
||||||
return json_response({"error": f"Pet {key} does not exist"}, status=404)
|
|
||||||
|
|
||||||
|
|
||||||
app = Application(middlewares=[pet_not_found_to_404])
|
|
||||||
oas.setup(app)
|
|
||||||
|
|
||||||
app["model"] = Model()
|
|
||||||
app.router.add_view("/pets", PetCollectionView)
|
|
||||||
app.router.add_view("/pets/{id}", PetItemView)
|
|
@@ -4,11 +4,6 @@ from pydantic import BaseModel
|
|||||||
class Pet(BaseModel):
|
class Pet(BaseModel):
|
||||||
id: int
|
id: int
|
||||||
name: str
|
name: str
|
||||||
age: int
|
|
||||||
|
|
||||||
|
|
||||||
class Error(BaseModel):
|
|
||||||
error: str
|
|
||||||
|
|
||||||
|
|
||||||
class Model:
|
class Model:
|
||||||
|
24
demo/view.py
24
demo/view.py
@@ -1,34 +1,28 @@
|
|||||||
from typing import List, Optional, Union
|
from aiohttp_pydantic import PydanticView
|
||||||
|
|
||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
|
|
||||||
from aiohttp_pydantic import PydanticView
|
from .model import Pet
|
||||||
from aiohttp_pydantic.oas.typing import r200, r201, r204, r404
|
|
||||||
|
|
||||||
from .model import Error, Pet
|
|
||||||
|
|
||||||
|
|
||||||
class PetCollectionView(PydanticView):
|
class PetCollectionView(PydanticView):
|
||||||
async def get(self, age: Optional[int] = None) -> r200[List[Pet]]:
|
async def get(self):
|
||||||
pets = self.request.app["model"].list_pets()
|
pets = self.request.app["model"].list_pets()
|
||||||
return web.json_response(
|
return web.json_response([pet.dict() for pet in pets])
|
||||||
[pet.dict() for pet in pets if age is None or age == pet.age]
|
|
||||||
)
|
|
||||||
|
|
||||||
async def post(self, pet: Pet) -> r201[Pet]:
|
async def post(self, pet: Pet):
|
||||||
self.request.app["model"].add_pet(pet)
|
self.request.app["model"].add_pet(pet)
|
||||||
return web.json_response(pet.dict())
|
return web.json_response(pet.dict())
|
||||||
|
|
||||||
|
|
||||||
class PetItemView(PydanticView):
|
class PetItemView(PydanticView):
|
||||||
async def get(self, id: int, /) -> Union[r200[Pet], r404[Error]]:
|
async def get(self, id: int, /):
|
||||||
pet = self.request.app["model"].find_pet(id)
|
pet = self.request.app["model"].find_pet(id)
|
||||||
return web.json_response(pet.dict())
|
return web.json_response(pet.dict())
|
||||||
|
|
||||||
async def put(self, id: int, /, pet: Pet) -> r200[Pet]:
|
async def put(self, id: int, /, pet: Pet):
|
||||||
self.request.app["model"].update_pet(id, pet)
|
self.request.app["model"].update_pet(id, pet)
|
||||||
return web.json_response(pet.dict())
|
return web.json_response(pet.dict())
|
||||||
|
|
||||||
async def delete(self, id: int, /) -> r204:
|
async def delete(self, id: int, /):
|
||||||
self.request.app["model"].remove_pet(id)
|
self.request.app["model"].remove_pet(id)
|
||||||
return web.Response(status=204)
|
return web.json_response(id)
|
||||||
|
@@ -1,27 +0,0 @@
|
|||||||
from aiohttp import web
|
|
||||||
|
|
||||||
from aiohttp_pydantic import PydanticView
|
|
||||||
|
|
||||||
|
|
||||||
class View1(PydanticView):
|
|
||||||
async def get(self, a: int, /):
|
|
||||||
return web.json_response()
|
|
||||||
|
|
||||||
|
|
||||||
class View2(PydanticView):
|
|
||||||
async def post(self, b: int, /):
|
|
||||||
return web.json_response()
|
|
||||||
|
|
||||||
|
|
||||||
sub_app = web.Application()
|
|
||||||
sub_app.router.add_view("/route-2/{b}", View2)
|
|
||||||
|
|
||||||
app = web.Application()
|
|
||||||
app.router.add_view("/route-1/{a}", View1)
|
|
||||||
app.add_subapp("/sub-app", sub_app)
|
|
||||||
|
|
||||||
|
|
||||||
def make_app():
|
|
||||||
app = web.Application()
|
|
||||||
app.router.add_view("/route-3/{a}", View1)
|
|
||||||
return app
|
|
@@ -1,120 +0,0 @@
|
|||||||
import argparse
|
|
||||||
from textwrap import dedent
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
from aiohttp_pydantic.oas import cmd
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def cmd_line():
|
|
||||||
parser = argparse.ArgumentParser()
|
|
||||||
cmd.setup(parser)
|
|
||||||
return parser
|
|
||||||
|
|
||||||
|
|
||||||
def test_show_oad_of_app(cmd_line, capfd):
|
|
||||||
args = cmd_line.parse_args(["tests.test_oas.test_cmd.sample"])
|
|
||||||
args.func(args)
|
|
||||||
captured = capfd.readouterr()
|
|
||||||
expected = dedent(
|
|
||||||
"""
|
|
||||||
{
|
|
||||||
"openapi": "3.0.0",
|
|
||||||
"paths": {
|
|
||||||
"/route-1/{a}": {
|
|
||||||
"get": {
|
|
||||||
"parameters": [
|
|
||||||
{
|
|
||||||
"in": "path",
|
|
||||||
"name": "a",
|
|
||||||
"required": true,
|
|
||||||
"schema": {
|
|
||||||
"type": "integer"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
]
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"/sub-app/route-2/{b}": {
|
|
||||||
"post": {
|
|
||||||
"parameters": [
|
|
||||||
{
|
|
||||||
"in": "path",
|
|
||||||
"name": "b",
|
|
||||||
"required": true,
|
|
||||||
"schema": {
|
|
||||||
"type": "integer"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
"""
|
|
||||||
)
|
|
||||||
|
|
||||||
assert captured.out.strip() == expected.strip()
|
|
||||||
|
|
||||||
|
|
||||||
def test_show_oad_of_sub_app(cmd_line, capfd):
|
|
||||||
args = cmd_line.parse_args(["tests.test_oas.test_cmd.sample:sub_app"])
|
|
||||||
args.func(args)
|
|
||||||
captured = capfd.readouterr()
|
|
||||||
expected = dedent(
|
|
||||||
"""
|
|
||||||
{
|
|
||||||
"openapi": "3.0.0",
|
|
||||||
"paths": {
|
|
||||||
"/sub-app/route-2/{b}": {
|
|
||||||
"post": {
|
|
||||||
"parameters": [
|
|
||||||
{
|
|
||||||
"in": "path",
|
|
||||||
"name": "b",
|
|
||||||
"required": true,
|
|
||||||
"schema": {
|
|
||||||
"type": "integer"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
"""
|
|
||||||
)
|
|
||||||
|
|
||||||
assert captured.out.strip() == expected.strip()
|
|
||||||
|
|
||||||
|
|
||||||
def test_show_oad_of_a_callable(cmd_line, capfd):
|
|
||||||
args = cmd_line.parse_args(["tests.test_oas.test_cmd.sample:make_app()"])
|
|
||||||
args.func(args)
|
|
||||||
captured = capfd.readouterr()
|
|
||||||
expected = dedent(
|
|
||||||
"""
|
|
||||||
{
|
|
||||||
"openapi": "3.0.0",
|
|
||||||
"paths": {
|
|
||||||
"/route-3/{a}": {
|
|
||||||
"get": {
|
|
||||||
"parameters": [
|
|
||||||
{
|
|
||||||
"in": "path",
|
|
||||||
"name": "a",
|
|
||||||
"required": true,
|
|
||||||
"schema": {
|
|
||||||
"type": "integer"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
"""
|
|
||||||
)
|
|
||||||
|
|
||||||
assert captured.out.strip() == expected.strip()
|
|
@@ -1,11 +1,8 @@
|
|||||||
from typing import List, Union
|
from pydantic.main import BaseModel
|
||||||
|
from aiohttp_pydantic import PydanticView, oas
|
||||||
|
from aiohttp import web
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from aiohttp import web
|
|
||||||
from pydantic.main import BaseModel
|
|
||||||
|
|
||||||
from aiohttp_pydantic import PydanticView, oas
|
|
||||||
from aiohttp_pydantic.oas.typing import r200, r201, r204, r404
|
|
||||||
|
|
||||||
|
|
||||||
class Pet(BaseModel):
|
class Pet(BaseModel):
|
||||||
@@ -14,21 +11,21 @@ class Pet(BaseModel):
|
|||||||
|
|
||||||
|
|
||||||
class PetCollectionView(PydanticView):
|
class PetCollectionView(PydanticView):
|
||||||
async def get(self) -> r200[List[Pet]]:
|
async def get(self):
|
||||||
return web.json_response()
|
return web.json_response()
|
||||||
|
|
||||||
async def post(self, pet: Pet) -> r201[Pet]:
|
async def post(self, pet: Pet):
|
||||||
return web.json_response()
|
return web.json_response()
|
||||||
|
|
||||||
|
|
||||||
class PetItemView(PydanticView):
|
class PetItemView(PydanticView):
|
||||||
async def get(self, id: int, /) -> Union[r200[Pet], r404]:
|
async def get(self, id: int, /):
|
||||||
return web.json_response()
|
return web.json_response()
|
||||||
|
|
||||||
async def put(self, id: int, /, pet: Pet):
|
async def put(self, id: int, /, pet: Pet):
|
||||||
return web.json_response()
|
return web.json_response()
|
||||||
|
|
||||||
async def delete(self, id: int, /) -> r204:
|
async def delete(self, id: int, /):
|
||||||
return web.json_response()
|
return web.json_response()
|
||||||
|
|
||||||
|
|
||||||
@@ -51,28 +48,7 @@ async def test_generated_oas_should_have_pets_paths(generated_oas):
|
|||||||
|
|
||||||
|
|
||||||
async def test_pets_route_should_have_get_method(generated_oas):
|
async def test_pets_route_should_have_get_method(generated_oas):
|
||||||
assert generated_oas["paths"]["/pets"]["get"] == {
|
assert generated_oas["paths"]["/pets"]["get"] == {}
|
||||||
"responses": {
|
|
||||||
"200": {
|
|
||||||
"content": {
|
|
||||||
"application/json": {
|
|
||||||
"schema": {
|
|
||||||
"type": "array",
|
|
||||||
"items": {
|
|
||||||
"title": "Pet",
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"id": {"title": "Id", "type": "integer"},
|
|
||||||
"name": {"title": "Name", "type": "string"},
|
|
||||||
},
|
|
||||||
"required": ["id", "name"],
|
|
||||||
},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
async def test_pets_route_should_have_post_method(generated_oas):
|
async def test_pets_route_should_have_post_method(generated_oas):
|
||||||
@@ -81,34 +57,17 @@ async def test_pets_route_should_have_post_method(generated_oas):
|
|||||||
"content": {
|
"content": {
|
||||||
"application/json": {
|
"application/json": {
|
||||||
"schema": {
|
"schema": {
|
||||||
"title": "Pet",
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
"properties": {
|
||||||
"id": {"title": "Id", "type": "integer"},
|
"id": {"title": "Id", "type": "integer"},
|
||||||
"name": {"title": "Name", "type": "string"},
|
"name": {"title": "Name", "type": "string"},
|
||||||
},
|
},
|
||||||
"required": ["id", "name"],
|
"required": ["id", "name"],
|
||||||
|
"title": "Pet",
|
||||||
|
"type": "object",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
},
|
}
|
||||||
"responses": {
|
|
||||||
"201": {
|
|
||||||
"content": {
|
|
||||||
"application/json": {
|
|
||||||
"schema": {
|
|
||||||
"title": "Pet",
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"id": {"title": "Id", "type": "integer"},
|
|
||||||
"name": {"title": "Name", "type": "string"},
|
|
||||||
},
|
|
||||||
"required": ["id", "name"],
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -120,13 +79,12 @@ async def test_pets_id_route_should_have_delete_method(generated_oas):
|
|||||||
assert generated_oas["paths"]["/pets/{id}"]["delete"] == {
|
assert generated_oas["paths"]["/pets/{id}"]["delete"] == {
|
||||||
"parameters": [
|
"parameters": [
|
||||||
{
|
{
|
||||||
"required": True,
|
|
||||||
"in": "path",
|
"in": "path",
|
||||||
"name": "id",
|
"name": "id",
|
||||||
|
"required": True,
|
||||||
"schema": {"type": "integer"},
|
"schema": {"type": "integer"},
|
||||||
}
|
}
|
||||||
],
|
]
|
||||||
"responses": {"204": {"content": {}}},
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -139,25 +97,7 @@ async def test_pets_id_route_should_have_get_method(generated_oas):
|
|||||||
"required": True,
|
"required": True,
|
||||||
"schema": {"type": "integer"},
|
"schema": {"type": "integer"},
|
||||||
}
|
}
|
||||||
],
|
]
|
||||||
"responses": {
|
|
||||||
"200": {
|
|
||||||
"content": {
|
|
||||||
"application/json": {
|
|
||||||
"schema": {
|
|
||||||
"properties": {
|
|
||||||
"id": {"title": "Id", "type": "integer"},
|
|
||||||
"name": {"title": "Name", "type": "string"},
|
|
||||||
},
|
|
||||||
"required": ["id", "name"],
|
|
||||||
"title": "Pet",
|
|
||||||
"type": "object",
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"404": {"content": {}},
|
|
||||||
},
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@@ -1,8 +1,6 @@
|
|||||||
from uuid import UUID
|
|
||||||
|
|
||||||
from pydantic import BaseModel
|
|
||||||
|
|
||||||
from aiohttp_pydantic.injectors import _parse_func_signature
|
from aiohttp_pydantic.injectors import _parse_func_signature
|
||||||
|
from pydantic import BaseModel
|
||||||
|
from uuid import UUID
|
||||||
|
|
||||||
|
|
||||||
class User(BaseModel):
|
class User(BaseModel):
|
||||||
|
@@ -1,8 +1,6 @@
|
|||||||
from typing import Optional
|
|
||||||
|
|
||||||
from aiohttp import web
|
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
|
from typing import Optional
|
||||||
|
from aiohttp import web
|
||||||
from aiohttp_pydantic import PydanticView
|
from aiohttp_pydantic import PydanticView
|
||||||
|
|
||||||
|
|
||||||
|
@@ -1,9 +1,7 @@
|
|||||||
import json
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
|
|
||||||
from aiohttp_pydantic import PydanticView
|
from aiohttp_pydantic import PydanticView
|
||||||
|
from datetime import datetime
|
||||||
|
import json
|
||||||
|
|
||||||
|
|
||||||
class JSONEncoder(json.JSONEncoder):
|
class JSONEncoder(json.JSONEncoder):
|
||||||
|
@@ -1,5 +1,4 @@
|
|||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
|
|
||||||
from aiohttp_pydantic import PydanticView
|
from aiohttp_pydantic import PydanticView
|
||||||
|
|
||||||
|
|
||||||
|
@@ -1,13 +1,10 @@
|
|||||||
from typing import Optional
|
|
||||||
|
|
||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
|
|
||||||
from aiohttp_pydantic import PydanticView
|
from aiohttp_pydantic import PydanticView
|
||||||
|
|
||||||
|
|
||||||
class ArticleView(PydanticView):
|
class ArticleView(PydanticView):
|
||||||
async def get(self, with_comments: bool, age: Optional[int] = None):
|
async def get(self, with_comments: bool):
|
||||||
return web.json_response({"with_comments": with_comments, "age": age})
|
return web.json_response({"with_comments": with_comments})
|
||||||
|
|
||||||
|
|
||||||
async def test_get_article_without_required_qs_should_return_an_error_message(
|
async def test_get_article_without_required_qs_should_return_an_error_message(
|
||||||
@@ -55,22 +52,7 @@ async def test_get_article_with_valid_qs_should_return_the_parsed_type(
|
|||||||
app.router.add_view("/article", ArticleView)
|
app.router.add_view("/article", ArticleView)
|
||||||
|
|
||||||
client = await aiohttp_client(app)
|
client = await aiohttp_client(app)
|
||||||
|
|
||||||
resp = await client.get("/article", params={"with_comments": "yes", "age": 3})
|
|
||||||
assert resp.status == 200
|
|
||||||
assert resp.content_type == "application/json"
|
|
||||||
assert await resp.json() == {"with_comments": True, "age": 3}
|
|
||||||
|
|
||||||
|
|
||||||
async def test_get_article_with_valid_qs_and_omitted_optional_should_return_none(
|
|
||||||
aiohttp_client, loop
|
|
||||||
):
|
|
||||||
app = web.Application()
|
|
||||||
app.router.add_view("/article", ArticleView)
|
|
||||||
|
|
||||||
client = await aiohttp_client(app)
|
|
||||||
|
|
||||||
resp = await client.get("/article", params={"with_comments": "yes"})
|
resp = await client.get("/article", params={"with_comments": "yes"})
|
||||||
assert resp.status == 200
|
assert resp.status == 200
|
||||||
assert resp.content_type == "application/json"
|
assert resp.content_type == "application/json"
|
||||||
assert await resp.json() == {"with_comments": True, "age": None}
|
assert await resp.json() == {"with_comments": True}
|
||||||
|
Reference in New Issue
Block a user