Compare commits

1 Commits

Author SHA1 Message Date
Vincent Maillol
2a064a75d9 update version 2020-10-25 20:52:04 +01:00
24 changed files with 103 additions and 633 deletions

View File

@@ -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

View File

@@ -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__")

View File

@@ -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

View File

@@ -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(

View File

@@ -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)

View File

@@ -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))

View File

@@ -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):

View File

@@ -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")

View File

@@ -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):

View File

@@ -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

View File

@@ -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):

View File

@@ -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)

View File

@@ -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)

View File

@@ -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:

View File

@@ -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)

View File

@@ -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

View File

@@ -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()

View File

@@ -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": {}},
},
} }

View File

@@ -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):

View File

@@ -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

View File

@@ -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):

View File

@@ -1,5 +1,4 @@
from aiohttp import web from aiohttp import web
from aiohttp_pydantic import PydanticView from aiohttp_pydantic import PydanticView

View File

@@ -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}