Compare commits
1 Commits
Author | SHA1 | Date | |
---|---|---|---|
|
2a064a75d9 |
67
README.rst
67
README.rst
@@ -145,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/
|
||||
.. _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
|
||||
reading annotation in your PydanticView. Use *aiohttp_pydantic.oas.setup()* to add the sub-application
|
||||
@@ -168,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')
|
||||
|
||||
If you want generate the Open Api Specification from specific aiohttp sub-applications.
|
||||
on the same route, you must use *apps_to_expose* parameter.
|
||||
If you want generate the Open Api Specification from several aiohttp sub-application.
|
||||
on the same route, you must use *apps_to_expose* parameters
|
||||
|
||||
|
||||
.. code-block:: python3
|
||||
@@ -179,61 +179,11 @@ on the same route, you must use *apps_to_expose* parameter.
|
||||
|
||||
app = web.Application()
|
||||
sub_app_1 = web.Application()
|
||||
sub_app_2 = web.Application()
|
||||
|
||||
oas.setup(app, apps_to_expose=[sub_app_1, sub_app_2])
|
||||
|
||||
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)
|
||||
oas.setup(app, apps_to_expose=[app, sub_app_1])
|
||||
|
||||
Demo
|
||||
----
|
||||
====
|
||||
|
||||
Have a look at `demo`_ for a complete example
|
||||
|
||||
@@ -246,11 +196,6 @@ Have a look at `demo`_ for a complete example
|
||||
|
||||
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
|
||||
|
@@ -1,5 +1,5 @@
|
||||
from .view import PydanticView
|
||||
|
||||
__version__ = "1.4.0"
|
||||
__version__ = "1.1.0"
|
||||
|
||||
__all__ = ("PydanticView", "__version__")
|
||||
|
@@ -1,9 +1,11 @@
|
||||
import abc
|
||||
from inspect import signature
|
||||
from typing import Callable, Tuple
|
||||
|
||||
from aiohttp.web_request import BaseRequest
|
||||
from pydantic import BaseModel
|
||||
from inspect import signature
|
||||
|
||||
|
||||
import abc
|
||||
|
||||
|
||||
class AbstractInjector(metaclass=abc.ABCMeta):
|
||||
|
@@ -1,11 +1,10 @@
|
||||
from importlib import resources
|
||||
from typing import Iterable
|
||||
from importlib import resources
|
||||
|
||||
import jinja2
|
||||
from aiohttp import web
|
||||
from swagger_ui_bundle import swagger_ui_path
|
||||
|
||||
from .view import get_oas, oas_ui
|
||||
from swagger_ui_bundle import swagger_ui_path
|
||||
|
||||
|
||||
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:
|
||||
def __init__(self, spec: dict):
|
||||
self._spec = spec.setdefault("info", {})
|
||||
@@ -118,39 +115,6 @@ class Parameters:
|
||||
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:
|
||||
def __init__(self, spec: dict):
|
||||
self._spec = spec
|
||||
@@ -179,10 +143,6 @@ class OperationObject:
|
||||
def parameters(self) -> Parameters:
|
||||
return Parameters(self._spec)
|
||||
|
||||
@property
|
||||
def responses(self) -> Responses:
|
||||
return Responses(self._spec)
|
||||
|
||||
|
||||
class PathItem:
|
||||
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,116 +1,51 @@
|
||||
import typing
|
||||
from typing import List, Type
|
||||
|
||||
from aiohttp.web import Response, json_response
|
||||
from aiohttp.web_app import Application
|
||||
from pydantic import BaseModel
|
||||
from aiohttp.web import json_response, Response
|
||||
|
||||
from aiohttp_pydantic.oas.struct import OpenApiSpec3, OperationObject, PathItem
|
||||
from typing import Type
|
||||
|
||||
from ..injectors import _parse_func_signature
|
||||
from ..view import PydanticView, is_pydantic_view
|
||||
from .typing import is_status_code_type
|
||||
|
||||
|
||||
JSON_SCHEMA_TYPES = {float: "number", str: "string", int: "integer"}
|
||||
|
||||
|
||||
def _is_pydantic_base_model(obj):
|
||||
"""
|
||||
Return true is obj is a pydantic.BaseModel subclass.
|
||||
"""
|
||||
try:
|
||||
return issubclass(obj, BaseModel)
|
||||
except TypeError:
|
||||
return False
|
||||
|
||||
|
||||
class _OASResponseBuilder:
|
||||
"""
|
||||
Parse the type annotated as returned by a function and
|
||||
generate the OAS operation response.
|
||||
"""
|
||||
|
||||
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)
|
||||
def _add_http_method_to_oas(oas_path: PathItem, method: str, view: Type[PydanticView]):
|
||||
method = method.lower()
|
||||
mtd: OperationObject = getattr(oas_path, method)
|
||||
handler = getattr(view, method)
|
||||
path_args, body_args, qs_args, header_args = _parse_func_signature(handler)
|
||||
|
||||
if body_args:
|
||||
oas_operation.request_body.content = {
|
||||
mtd.request_body.content = {
|
||||
"application/json": {"schema": next(iter(body_args.values())).schema()}
|
||||
}
|
||||
|
||||
i = 0
|
||||
for i, (name, type_) in enumerate(path_args.items()):
|
||||
oas_operation.parameters[i].required = True
|
||||
oas_operation.parameters[i].in_ = "path"
|
||||
oas_operation.parameters[i].name = name
|
||||
oas_operation.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
||||
mtd.parameters[i].required = True
|
||||
mtd.parameters[i].in_ = "path"
|
||||
mtd.parameters[i].name = name
|
||||
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
||||
|
||||
for i, (name, type_) in enumerate(qs_args.items(), i + 1):
|
||||
oas_operation.parameters[i].required = False
|
||||
oas_operation.parameters[i].in_ = "query"
|
||||
oas_operation.parameters[i].name = name
|
||||
oas_operation.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
||||
mtd.parameters[i].required = False
|
||||
mtd.parameters[i].in_ = "query"
|
||||
mtd.parameters[i].name = name
|
||||
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
|
||||
|
||||
for i, (name, type_) in enumerate(header_args.items(), i + 1):
|
||||
oas_operation.parameters[i].required = False
|
||||
oas_operation.parameters[i].in_ = "header"
|
||||
oas_operation.parameters[i].name = name
|
||||
oas_operation.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)
|
||||
mtd.parameters[i].required = False
|
||||
mtd.parameters[i].in_ = "header"
|
||||
mtd.parameters[i].name = name
|
||||
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[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()
|
||||
for app in apps:
|
||||
for resources in app.router.resources():
|
||||
@@ -125,15 +60,7 @@ def generate_oas(apps: List[Application]) -> dict:
|
||||
else:
|
||||
_add_http_method_to_oas(path, resource_route.method, view)
|
||||
|
||||
return 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))
|
||||
return json_response(oas.spec)
|
||||
|
||||
|
||||
async def oas_ui(request):
|
||||
|
@@ -1,20 +1,20 @@
|
||||
from functools import update_wrapper
|
||||
from inspect import iscoroutinefunction
|
||||
from typing import Any, Callable, Generator, Iterable
|
||||
|
||||
from aiohttp.abc import AbstractView
|
||||
from aiohttp.hdrs import METH_ALL
|
||||
from aiohttp.web import json_response
|
||||
from aiohttp.web_exceptions import HTTPMethodNotAllowed
|
||||
from aiohttp.web_response import StreamResponse
|
||||
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,
|
||||
HeadersGetter,
|
||||
QueryGetter,
|
||||
BodyGetter,
|
||||
AbstractInjector,
|
||||
_parse_func_signature,
|
||||
)
|
||||
|
||||
|
@@ -1,5 +1,25 @@
|
||||
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)
|
||||
|
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)
|
@@ -6,10 +6,6 @@ class Pet(BaseModel):
|
||||
name: str
|
||||
|
||||
|
||||
class Error(BaseModel):
|
||||
error: str
|
||||
|
||||
|
||||
class Model:
|
||||
"""
|
||||
To keep simple this demo, we use a simple dict as database to
|
||||
|
20
demo/view.py
20
demo/view.py
@@ -1,32 +1,28 @@
|
||||
from typing import List, Union
|
||||
|
||||
from aiohttp_pydantic import PydanticView
|
||||
from aiohttp import web
|
||||
|
||||
from aiohttp_pydantic import PydanticView
|
||||
from aiohttp_pydantic.oas.typing import r200, r201, r204, r404
|
||||
|
||||
from .model import Error, Pet
|
||||
from .model import Pet
|
||||
|
||||
|
||||
class PetCollectionView(PydanticView):
|
||||
async def get(self) -> r200[List[Pet]]:
|
||||
async def get(self):
|
||||
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]:
|
||||
async def post(self, pet: 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]]:
|
||||
async def get(self, id: int, /):
|
||||
pet = self.request.app["model"].find_pet(id)
|
||||
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)
|
||||
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)
|
||||
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
|
||||
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):
|
||||
@@ -14,21 +11,21 @@ class Pet(BaseModel):
|
||||
|
||||
|
||||
class PetCollectionView(PydanticView):
|
||||
async def get(self) -> r200[List[Pet]]:
|
||||
async def get(self):
|
||||
return web.json_response()
|
||||
|
||||
async def post(self, pet: Pet) -> r201[Pet]:
|
||||
async def post(self, pet: Pet):
|
||||
return web.json_response()
|
||||
|
||||
|
||||
class PetItemView(PydanticView):
|
||||
async def get(self, id: int, /) -> Union[r200[Pet], r404]:
|
||||
async def get(self, id: int, /):
|
||||
return web.json_response()
|
||||
|
||||
async def put(self, id: int, /, pet: Pet):
|
||||
return web.json_response()
|
||||
|
||||
async def delete(self, id: int, /) -> r204:
|
||||
async def delete(self, id: int, /):
|
||||
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):
|
||||
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"],
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
assert generated_oas["paths"]["/pets"]["get"] == {}
|
||||
|
||||
|
||||
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": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"title": "Pet",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {"title": "Id", "type": "integer"},
|
||||
"name": {"title": "Name", "type": "string"},
|
||||
},
|
||||
"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"] == {
|
||||
"parameters": [
|
||||
{
|
||||
"required": True,
|
||||
"in": "path",
|
||||
"name": "id",
|
||||
"required": True,
|
||||
"schema": {"type": "integer"},
|
||||
}
|
||||
],
|
||||
"responses": {"204": {"content": {}}},
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@@ -139,25 +97,7 @@ async def test_pets_id_route_should_have_get_method(generated_oas):
|
||||
"required": True,
|
||||
"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 pydantic import BaseModel
|
||||
from uuid import UUID
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
|
@@ -1,8 +1,6 @@
|
||||
from typing import Optional
|
||||
|
||||
from aiohttp import web
|
||||
from pydantic import BaseModel
|
||||
|
||||
from typing import Optional
|
||||
from aiohttp import web
|
||||
from aiohttp_pydantic import PydanticView
|
||||
|
||||
|
||||
|
@@ -1,9 +1,7 @@
|
||||
import json
|
||||
from datetime import datetime
|
||||
|
||||
from aiohttp import web
|
||||
|
||||
from aiohttp_pydantic import PydanticView
|
||||
from datetime import datetime
|
||||
import json
|
||||
|
||||
|
||||
class JSONEncoder(json.JSONEncoder):
|
||||
|
@@ -1,5 +1,4 @@
|
||||
from aiohttp import web
|
||||
|
||||
from aiohttp_pydantic import PydanticView
|
||||
|
||||
|
||||
|
@@ -1,5 +1,4 @@
|
||||
from aiohttp import web
|
||||
|
||||
from aiohttp_pydantic import PydanticView
|
||||
|
||||
|
||||
|
Reference in New Issue
Block a user