Compare commits

1 Commits

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

3
.gitignore vendored
View File

@@ -1,9 +1,6 @@
.coverage
.idea/
.pytest_cache
__pycache__
aiohttp_pydantic.egg-info/
build/
coverage.xml
dist/

View File

@@ -2,14 +2,11 @@ language: python
python:
- '3.8'
script:
- pytest --cov-report=xml --cov=aiohttp_pydantic tests/
- pytest tests/
install:
- pip install -U setuptools wheel pip
- pip install -r requirements/test.txt
- pip install -r requirements/ci.txt
- pip install -r test_requirements.txt
- pip install .
after_success:
- codecov
deploy:
provider: pypi
username: __token__
@@ -19,4 +16,4 @@ deploy:
on:
tags: true
branch: main
python: '3.8'
python: '3.8'

View File

@@ -1,28 +1,6 @@
Aiohttp pydantic - Aiohttp View to validate and parse request
=============================================================
.. image:: https://travis-ci.org/Maillol/aiohttp-pydantic.svg?branch=main
:target: https://travis-ci.org/Maillol/aiohttp-pydantic
.. image:: https://img.shields.io/pypi/v/aiohttp-pydantic
:target: https://img.shields.io/pypi/v/aiohttp-pydantic
:alt: Latest PyPI package version
.. image:: https://codecov.io/gh/Maillol/aiohttp-pydantic/branch/main/graph/badge.svg
:target: https://codecov.io/gh/Maillol/aiohttp-pydantic
:alt: codecov.io status for master branch
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
--------------
@@ -68,7 +46,6 @@ Example:
$ curl -X GET http://127.0.0.1:8080/article?with_comments=a
[
{
"in": "query string",
"loc": [
"with_comments"
],
@@ -83,7 +60,6 @@ Example:
$ curl -H "Content-Type: application/json" -X post http://127.0.0.1:8080/article --data '{}'
[
{
"in": "body",
"loc": [
"name"
],
@@ -118,7 +94,7 @@ Example:
Inject Query String Parameters
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
To declare a query parameters, you must declare your argument as a simple argument:
To declare a query parameters, you must declare your argument as simple argument:
.. code-block:: python3
@@ -133,7 +109,7 @@ To declare a query parameters, you must declare your argument as a simple argume
Inject Request Body
~~~~~~~~~~~~~~~~~~~
To declare a body parameter, you must declare your argument as a simple argument annotated with `pydantic Model`_.
To declare a body parameters, you must declare your argument as a simple argument annotated with `pydantic Model`_.
.. code-block:: python3
@@ -169,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
@@ -192,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
@@ -203,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
@@ -270,12 +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
.. _aiohttp view: https://docs.aiohttp.org/en/stable/web_quickstart.html#class-based-views

View File

@@ -1,5 +1,5 @@
from .view import PydanticView
__version__ = "1.6.0"
__version__ = "1.1.0"
__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 aiohttp.web_exceptions import HTTPBadRequest
from aiohttp.web_request import BaseRequest
from pydantic import BaseModel
from inspect import signature
from .utils import is_pydantic_base_model
import abc
class AbstractInjector(metaclass=abc.ABCMeta):
@@ -15,16 +13,8 @@ class AbstractInjector(metaclass=abc.ABCMeta):
An injector parse HTTP request and inject params to the view.
"""
@property
@abc.abstractmethod
def context(self) -> str:
"""
The name of part of parsed request
i.e "HTTP header", "URL path", ...
"""
@abc.abstractmethod
def __init__(self, args_spec: dict, default_values: dict):
def __init__(self, args_spec: dict):
"""
args_spec - ordered mapping: arg_name -> type
"""
@@ -41,12 +31,8 @@ class MatchInfoGetter(AbstractInjector):
Validates and injects the part of URL path inside the view positional args.
"""
context = "path"
def __init__(self, args_spec: dict, default_values: dict):
attrs = {"__annotations__": args_spec}
attrs.update(default_values)
self.model = type("PathModel", (BaseModel,), attrs)
def __init__(self, args_spec: dict):
self.model = type("PathModel", (BaseModel,), {"__annotations__": args_spec})
def inject(self, request: BaseRequest, args_view: list, kwargs_view: dict):
args_view.extend(self.model(**request.match_info).dict().values())
@@ -57,19 +43,11 @@ class BodyGetter(AbstractInjector):
Validates and injects the content of request body inside the view kwargs.
"""
context = "body"
def __init__(self, args_spec: dict, default_values: dict):
def __init__(self, args_spec: dict):
self.arg_name, self.model = next(iter(args_spec.items()))
async def inject(self, request: BaseRequest, args_view: list, kwargs_view: dict):
try:
body = await request.json()
except JSONDecodeError:
raise HTTPBadRequest(
text='{"error": "Malformed JSON"}', content_type="application/json"
)
body = await request.json()
kwargs_view[self.arg_name] = self.model(**body)
@@ -78,12 +56,8 @@ class QueryGetter(AbstractInjector):
Validates and injects the query string inside the view kwargs.
"""
context = "query string"
def __init__(self, args_spec: dict, default_values: dict):
attrs = {"__annotations__": args_spec}
attrs.update(default_values)
self.model = type("QueryModel", (BaseModel,), attrs)
def __init__(self, args_spec: dict):
self.model = type("QueryModel", (BaseModel,), {"__annotations__": args_spec})
def inject(self, request: BaseRequest, args_view: list, kwargs_view: dict):
kwargs_view.update(self.model(**request.query).dict())
@@ -94,33 +68,27 @@ class HeadersGetter(AbstractInjector):
Validates and injects the HTTP headers inside the view kwargs.
"""
context = "headers"
def __init__(self, args_spec: dict, default_values: dict):
attrs = {"__annotations__": args_spec}
attrs.update(default_values)
self.model = type("HeaderModel", (BaseModel,), attrs)
def __init__(self, args_spec: dict):
self.model = type("HeaderModel", (BaseModel,), {"__annotations__": args_spec})
def inject(self, request: BaseRequest, args_view: list, kwargs_view: dict):
header = {k.lower().replace("-", "_"): v for k, v in request.headers.items()}
kwargs_view.update(self.model(**header).dict())
def _parse_func_signature(func: Callable) -> Tuple[dict, dict, dict, dict, dict]:
def _parse_func_signature(func: Callable) -> Tuple[dict, dict, dict, dict]:
"""
Analyse function signature and returns 4-tuple:
0 - arguments will be set from the url path
1 - argument will be set from the request body.
2 - argument will be set from the query string.
3 - argument will be set from the HTTP headers.
4 - Default value for each parameters
"""
path_args = {}
body_args = {}
qs_args = {}
header_args = {}
defaults = {}
for param_name, param_spec in signature(func).parameters.items():
if param_name == "self":
@@ -129,14 +97,10 @@ def _parse_func_signature(func: Callable) -> Tuple[dict, dict, dict, dict, dict]
if param_spec.annotation == param_spec.empty:
raise RuntimeError(f"The parameter {param_name} must have an annotation")
if param_spec.default is not param_spec.empty:
defaults[param_name] = param_spec.default
if param_spec.kind is param_spec.POSITIONAL_ONLY:
path_args[param_name] = param_spec.annotation
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
else:
qs_args[param_name] = param_spec.annotation
@@ -145,4 +109,4 @@ def _parse_func_signature(func: Callable) -> Tuple[dict, dict, dict, dict, dict]
else:
raise RuntimeError(f"You cannot use {param_spec.VAR_POSITIONAL} parameters")
return path_args, body_args, qs_args, header_args, defaults
return path_args, body_args, qs_args, header_args

View File

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

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,17 +1,10 @@
"""
Utility to write Open Api Specifications using the Python language.
"""
from typing import Union
class Info:
def __init__(self, spec: dict):
self._spec = spec.setdefault("info", {})
@property
def title(self):
return self._spec.get("title")
return self._spec["title"]
@title.setter
def title(self, title):
@@ -19,7 +12,7 @@ class Info:
@property
def description(self):
return self._spec.get("description")
return self._spec["description"]
@description.setter
def description(self, description):
@@ -27,20 +20,12 @@ class Info:
@property
def version(self):
return self._spec.get("version")
return self._spec["version"]
@version.setter
def version(self, version):
self._spec["version"] = version
@property
def terms_of_service(self):
return self._spec.get("termsOfService")
@terms_of_service.setter
def terms_of_service(self, terms_of_service):
self._spec["termsOfService"] = terms_of_service
class RequestBody:
def __init__(self, spec: dict):
@@ -55,8 +40,8 @@ class RequestBody:
self._spec["description"] = description
@property
def required(self) -> bool:
return self._spec.get("required", False)
def required(self):
return self._spec["required"]
@required.setter
def required(self, required: bool):
@@ -130,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
@@ -191,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):
@@ -232,22 +180,6 @@ class PathItem:
def trace(self) -> OperationObject:
return OperationObject(self._spec.setdefault("trace", {}))
@property
def description(self) -> str:
return self._spec["description"]
@description.setter
def description(self, description: str):
self._spec["description"] = description
@property
def summary(self) -> str:
return self._spec["summary"]
@summary.setter
def summary(self, summary: str):
self._spec["summary"] = summary
class Paths:
def __init__(self, spec: dict):
@@ -272,7 +204,7 @@ class Server:
@property
def description(self) -> str:
return self._spec["description"]
return self._spec["url"]
@description.setter
def description(self, description: str):

View File

@@ -1,48 +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],))
return None
def is_status_code_type(obj) -> bool:
"""
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,166 +1,66 @@
import typing
from datetime import date, datetime
from inspect import getdoc
from itertools import count
from typing import List, Type
from uuid import UUID
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 ..utils import is_pydantic_base_model
from ..view import PydanticView, is_pydantic_view
from .typing import is_status_code_type
JSON_SCHEMA_TYPES = {
float: {"type": "number"},
str: {"type": "string"},
int: {"type": "integer"},
UUID: {"type": "string", "format": "uuid"},
bool: {"type": "boolean"},
datetime: {"type": "string", "format": "date-time"},
date: {"type": "string", "format": "date"},
}
def _handle_optional(type_):
"""
Returns the type wrapped in Optional or None.
>>> _handle_optional(int)
>>> _handle_optional(Optional[str])
<class 'str'>
"""
if typing.get_origin(type_) is typing.Union:
args = typing.get_args(type_)
if len(args) == 2 and type(None) in args:
return next(iter(set(args) - {type(None)}))
return None
JSON_SCHEMA_TYPES = {float: "number", str: "string", int: "integer"}
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)
path_args, body_args, qs_args, header_args, defaults = _parse_func_signature(
handler
)
description = getdoc(handler)
if description:
oas_operation.description = description
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()}
}
indexes = count()
for args_location, args in (
("path", path_args.items()),
("query", qs_args.items()),
("header", header_args.items()),
):
for name, type_ in args:
i = next(indexes)
oas_operation.parameters[i].in_ = args_location
oas_operation.parameters[i].name = name
optional_type = _handle_optional(type_)
i = 0
for i, (name, type_) in enumerate(path_args.items()):
mtd.parameters[i].required = True
mtd.parameters[i].in_ = "path"
mtd.parameters[i].name = name
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
attrs = {"__annotations__": {"__root__": type_}}
if name in defaults:
attrs["__root__"] = defaults[name]
for i, (name, type_) in enumerate(qs_args.items(), i + 1):
mtd.parameters[i].required = False
mtd.parameters[i].in_ = "query"
mtd.parameters[i].name = name
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
oas_operation.parameters[i].schema = type(
name, (BaseModel,), attrs
).schema()
oas_operation.parameters[i].required = optional_type is None
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:
"""
Generate and return Open Api Specification from PydanticView in application.
"""
oas = OpenApiSpec3()
for app in apps:
for resources in app.router.resources():
for resource_route in resources:
if not is_pydantic_view(resource_route.handler):
continue
view: Type[PydanticView] = resource_route.handler
info = resource_route.get_info()
path = oas.paths[info.get("path", info.get("formatter"))]
if resource_route.method == "*":
for method_name in view.allowed_methods:
_add_http_method_to_oas(path, method_name, view)
else:
_add_http_method_to_oas(path, resource_route.method, view)
return oas.spec
for i, (name, type_) in enumerate(header_args.items(), i + 1):
mtd.parameters[i].required = False
mtd.parameters[i].in_ = "header"
mtd.parameters[i].name = name
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
async def get_oas(request):
"""
View to generate the Open Api Specification from PydanticView in application.
Generate Open Api Specification from PydanticView in application.
"""
apps = request.app["apps to expose"]
return json_response(generate_oas(apps))
oas = OpenApiSpec3()
for app in apps:
for resources in app.router.resources():
for resource_route in resources:
if is_pydantic_view(resource_route.handler):
view: Type[PydanticView] = resource_route.handler
info = resource_route.get_info()
path = oas.paths[info.get("path", info.get("formatter"))]
if resource_route.method == "*":
for method_name in view.allowed_methods:
_add_http_method_to_oas(path, method_name, view)
else:
_add_http_method_to_oas(path, resource_route.method, view)
return json_response(oas.spec)
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 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, QueryGetter, _parse_func_signature)
from .injectors import (
MatchInfoGetter,
HeadersGetter,
QueryGetter,
BodyGetter,
AbstractInjector,
_parse_func_signature,
)
class PydanticView(AbstractView):
@@ -44,25 +50,16 @@ class PydanticView(AbstractView):
@staticmethod
def parse_func_signature(func: Callable) -> Iterable[AbstractInjector]:
path_args, body_args, qs_args, header_args, defaults = _parse_func_signature(
func
)
path_args, body_args, qs_args, header_args = _parse_func_signature(func)
injectors = []
def default_value(args: dict) -> dict:
"""
Returns the default values of args.
"""
return {name: defaults[name] for name in args if name in defaults}
if path_args:
injectors.append(MatchInfoGetter(path_args, default_value(path_args)))
injectors.append(MatchInfoGetter(path_args))
if body_args:
injectors.append(BodyGetter(body_args, default_value(body_args)))
injectors.append(BodyGetter(body_args))
if qs_args:
injectors.append(QueryGetter(qs_args, default_value(qs_args)))
injectors.append(QueryGetter(qs_args))
if header_args:
injectors.append(HeadersGetter(header_args, default_value(header_args)))
injectors.append(HeadersGetter(header_args))
return injectors
@@ -86,11 +83,7 @@ def inject_params(
else:
injector.inject(self.request, args, kwargs)
except ValidationError as error:
errors = error.errors()
for error in errors:
error["in"] = injector.context
return json_response(data=errors, status=400)
return json_response(text=error.json(), status=400)
return await handler(self, *args, **kwargs)

View File

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

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):
id: int
name: str
age: int
class Error(BaseModel):
error: str
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_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, age: Optional[int] = None) -> r200[List[Pet]]:
async def get(self):
pets = self.request.app["model"].list_pets()
return web.json_response(
[pet.dict() for pet in pets if age is None or age == pet.age]
)
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)

View File

@@ -1,7 +0,0 @@
certifi==2020.11.8
chardet==3.0.4
codecov==2.1.10
coverage==5.3
idna==2.10
requests==2.25.0
urllib3==1.26.2

View File

@@ -1,13 +0,0 @@
attrs==20.3.0
coverage==5.3
iniconfig==1.1.1
packaging==20.4
pluggy==0.13.1
py==1.9.0
pyparsing==2.4.7
pytest==6.1.2
pytest-aiohttp==0.3.0
pytest-cov==2.10.1
six==1.15.0
toml==0.10.2
typing-extensions==3.7.4.3

View File

@@ -35,8 +35,8 @@ install_requires =
swagger-ui-bundle
[options.extras_require]
test = pytest==6.1.2; pytest-aiohttp==0.3.0; pytest-cov==2.10.1
ci = pytest==6.1.2; pytest-aiohttp==0.3.0; pytest-cov==2.10.1; codecov==2.1.10
test = pytest; pytest-aiohttp
[options.packages.find]
exclude =

3
test_requirements.txt Normal file
View File

@@ -0,0 +1,3 @@
pytest==6.1.1
pytest-aiohttp==0.3.0
typing_extensions>=3.6.5

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,124 +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": {
"title": "a",
"type": "integer"
}
}
]
}
},
"/sub-app/route-2/{b}": {
"post": {
"parameters": [
{
"in": "path",
"name": "b",
"required": true,
"schema": {
"title": "b",
"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": {
"title": "b",
"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": {
"title": "a",
"type": "integer"
}
}
]
}
}
}
}
"""
)
assert captured.out.strip() == expected.strip()

View File

@@ -1,54 +0,0 @@
import pytest
from aiohttp_pydantic.oas.struct import OpenApiSpec3
def test_info_title():
oas = OpenApiSpec3()
assert oas.info.title is None
oas.info.title = "Info Title"
assert oas.info.title == "Info Title"
assert oas.spec == {"info": {"title": "Info Title"}, "openapi": "3.0.0"}
def test_info_description():
oas = OpenApiSpec3()
assert oas.info.description is None
oas.info.description = "info description"
assert oas.info.description == "info description"
assert oas.spec == {"info": {"description": "info description"}, "openapi": "3.0.0"}
def test_info_version():
oas = OpenApiSpec3()
assert oas.info.version is None
oas.info.version = "3.14"
assert oas.info.version == "3.14"
assert oas.spec == {"info": {"version": "3.14"}, "openapi": "3.0.0"}
def test_info_terms_of_service():
oas = OpenApiSpec3()
assert oas.info.terms_of_service is None
oas.info.terms_of_service = "http://example.com/terms/"
assert oas.info.terms_of_service == "http://example.com/terms/"
assert oas.spec == {
"info": {"termsOfService": "http://example.com/terms/"},
"openapi": "3.0.0",
}
@pytest.mark.skip("Not yet implemented")
def test_info_license():
oas = OpenApiSpec3()
oas.info.license.name = "Apache 2.0"
oas.info.license.url = "https://www.apache.org/licenses/LICENSE-2.0.html"
assert oas.spec == {
"info": {
"license": {
"name": "Apache 2.0",
"url": "https://www.apache.org/licenses/LICENSE-2.0.html",
}
},
"openapi": "3.0.0",
}

View File

@@ -1,124 +0,0 @@
from aiohttp_pydantic.oas.struct import OpenApiSpec3
def test_paths_description():
oas = OpenApiSpec3()
oas.paths["/users/{id}"].description = "This route ..."
assert oas.spec == {
"openapi": "3.0.0",
"paths": {"/users/{id}": {"description": "This route ..."}},
}
def test_paths_get():
oas = OpenApiSpec3()
oas.paths["/users/{id}"].get
assert oas.spec == {"openapi": "3.0.0", "paths": {"/users/{id}": {"get": {}}}}
def test_paths_operation_description():
oas = OpenApiSpec3()
operation = oas.paths["/users/{id}"].get
operation.description = "Long descriptions ..."
assert oas.spec == {
"openapi": "3.0.0",
"paths": {"/users/{id}": {"get": {"description": "Long descriptions ..."}}},
}
def test_paths_operation_summary():
oas = OpenApiSpec3()
operation = oas.paths["/users/{id}"].get
operation.summary = "Updates a pet in the store with form data"
assert oas.spec == {
"openapi": "3.0.0",
"paths": {
"/users/{id}": {
"get": {"summary": "Updates a pet in the store with form data"}
}
},
}
def test_paths_operation_parameters():
oas = OpenApiSpec3()
operation = oas.paths["/users/{petId}"].get
parameter = operation.parameters[0]
parameter.name = "petId"
parameter.description = "ID of pet that needs to be updated"
parameter.in_ = "path"
parameter.required = True
assert oas.spec == {
"openapi": "3.0.0",
"paths": {
"/users/{petId}": {
"get": {
"parameters": [
{
"description": "ID of pet that needs to be updated",
"in": "path",
"name": "petId",
"required": True,
}
]
}
}
},
}
def test_paths_operation_requestBody():
oas = OpenApiSpec3()
request_body = oas.paths["/users/{petId}"].get.request_body
request_body.description = "user to add to the system"
request_body.content = {
"application/json": {
"schema": {"$ref": "#/components/schemas/User"},
"examples": {
"user": {
"summary": "User Example",
"externalValue": "http://foo.bar/examples/user-example.json",
}
},
}
}
request_body.required = True
assert oas.spec == {
"openapi": "3.0.0",
"paths": {
"/users/{petId}": {
"get": {
"requestBody": {
"content": {
"application/json": {
"examples": {
"user": {
"externalValue": "http://foo.bar/examples/user-example.json",
"summary": "User Example",
}
},
"schema": {"$ref": "#/components/schemas/User"},
}
},
"description": "user to add to the system",
"required": True,
}
}
}
},
}
def test_paths_operation_responses():
oas = OpenApiSpec3()
response = oas.paths["/users/{petId}"].get.responses[200]
response.description = "A complex object array response"
response.content = {
"application/json": {
"schema": {
"type": "array",
"items": {"$ref": "#/components/schemas/VeryComplexType"},
}
}
}

View File

@@ -1,36 +0,0 @@
import pytest
from aiohttp_pydantic.oas.struct import OpenApiSpec3
def test_sever_url():
oas = OpenApiSpec3()
oas.servers[0].url = "https://development.gigantic-server.com/v1"
oas.servers[1].url = "https://development.gigantic-server.com/v2"
assert oas.spec == {
"openapi": "3.0.0",
"servers": [
{"url": "https://development.gigantic-server.com/v1"},
{"url": "https://development.gigantic-server.com/v2"},
],
}
def test_sever_description():
oas = OpenApiSpec3()
oas.servers[0].url = "https://development.gigantic-server.com/v1"
oas.servers[0].description = "Development server"
assert oas.spec == {
"openapi": "3.0.0",
"servers": [
{
"url": "https://development.gigantic-server.com/v1",
"description": "Development server",
}
],
}
@pytest.mark.skip("Not yet implemented")
def test_sever_variables():
oas = OpenApiSpec3()

View File

@@ -1,12 +1,8 @@
from typing import List, Optional, Union
from uuid import UUID
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):
@@ -15,27 +11,21 @@ class Pet(BaseModel):
class PetCollectionView(PydanticView):
async def get(
self, format: str, name: Optional[str] = None, *, promo: Optional[UUID] = None
) -> r200[List[Pet]]:
"""
Get a list of pets
"""
async def get(self):
return web.json_response()
async def post(self, pet: Pet) -> r201[Pet]:
"""Create a 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()
@@ -58,86 +48,26 @@ 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"] == {
"description": "Get a list of pets",
"parameters": [
{
"in": "query",
"name": "format",
"required": True,
"schema": {"title": "format", "type": "string"},
},
{
"in": "query",
"name": "name",
"required": False,
"schema": {"title": "name", "type": "string"},
},
{
"in": "header",
"name": "promo",
"required": False,
"schema": {"title": "promo", "format": "uuid", "type": "string"},
},
],
"responses": {
"200": {
"content": {
"application/json": {
"schema": {
"items": {
"properties": {
"id": {"title": "Id", "type": "integer"},
"name": {"title": "Name", "type": "string"},
},
"required": ["id", "name"],
"title": "Pet",
"type": "object",
},
"type": "array",
}
}
}
}
},
}
assert generated_oas["paths"]["/pets"]["get"] == {}
async def test_pets_route_should_have_post_method(generated_oas):
assert generated_oas["paths"]["/pets"]["post"] == {
"description": "Create a Pet",
"requestBody": {
"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"],
}
}
}
}
},
}
}
@@ -149,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",
"schema": {"title": "id", "type": "integer"},
"required": True,
"schema": {"type": "integer"},
}
],
"responses": {"204": {"content": {}}},
]
}
@@ -166,27 +95,9 @@ async def test_pets_id_route_should_have_get_method(generated_oas):
"in": "path",
"name": "id",
"required": True,
"schema": {"title": "id", "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": {}},
},
]
}
@@ -197,7 +108,7 @@ async def test_pets_id_route_should_have_put_method(generated_oas):
"in": "path",
"name": "id",
"required": True,
"schema": {"title": "id", "type": "integer"},
"schema": {"type": "integer"},
}
],
"requestBody": {

View File

@@ -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):
@@ -38,42 +36,32 @@ def test_parse_func_signature():
def path_body_qs_and_header(self, id: str, /, user: User, page: int, *, auth: UUID):
pass
assert _parse_func_signature(body_only) == ({}, {"user": User}, {}, {}, {})
assert _parse_func_signature(path_only) == ({"id": str}, {}, {}, {}, {})
assert _parse_func_signature(qs_only) == ({}, {}, {"page": int}, {}, {})
assert _parse_func_signature(header_only) == ({}, {}, {}, {"auth": UUID}, {})
assert _parse_func_signature(path_and_qs) == (
{"id": str},
{},
{"page": int},
{},
{},
)
assert _parse_func_signature(body_only) == ({}, {"user": User}, {}, {})
assert _parse_func_signature(path_only) == ({"id": str}, {}, {}, {})
assert _parse_func_signature(qs_only) == ({}, {}, {"page": int}, {})
assert _parse_func_signature(header_only) == ({}, {}, {}, {"auth": UUID})
assert _parse_func_signature(path_and_qs) == ({"id": str}, {}, {"page": int}, {})
assert _parse_func_signature(path_and_header) == (
{"id": str},
{},
{},
{"auth": UUID},
{},
)
assert _parse_func_signature(qs_and_header) == (
{},
{},
{"page": int},
{"auth": UUID},
{},
)
assert _parse_func_signature(path_qs_and_header) == (
{"id": str},
{},
{"page": int},
{"auth": UUID},
{},
)
assert _parse_func_signature(path_body_qs_and_header) == (
{"id": str},
{"user": User},
{"page": int},
{"auth": UUID},
{},
)

View File

@@ -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
@@ -27,12 +25,7 @@ async def test_post_an_article_without_required_field_should_return_an_error_mes
assert resp.status == 400
assert resp.content_type == "application/json"
assert await resp.json() == [
{
"in": "body",
"loc": ["name"],
"msg": "field required",
"type": "value_error.missing",
}
{"loc": ["name"], "msg": "field required", "type": "value_error.missing"}
]
@@ -48,7 +41,6 @@ async def test_post_an_article_with_wrong_type_field_should_return_an_error_mess
assert resp.content_type == "application/json"
assert await resp.json() == [
{
"in": "body",
"loc": ["nb_page"],
"msg": "value is not a valid integer",
"type": "type_error.integer",

View File

@@ -1,10 +1,7 @@
import json
from datetime import datetime
from enum import Enum
from aiohttp import web
from aiohttp_pydantic import PydanticView
from datetime import datetime
import json
class JSONEncoder(json.JSONEncoder):
@@ -22,16 +19,6 @@ class ArticleView(PydanticView):
)
class FormatEnum(str, Enum):
UTM = "UMT"
MGRS = "MGRS"
class ViewWithEnumType(PydanticView):
async def get(self, *, format: FormatEnum):
return web.json_response({"format": format}, dumps=JSONEncoder().encode)
async def test_get_article_without_required_header_should_return_an_error_message(
aiohttp_client, loop
):
@@ -44,7 +31,6 @@ async def test_get_article_without_required_header_should_return_an_error_messag
assert resp.content_type == "application/json"
assert await resp.json() == [
{
"in": "headers",
"loc": ["signature_expired"],
"msg": "field required",
"type": "value_error.missing",
@@ -64,7 +50,6 @@ async def test_get_article_with_wrong_header_type_should_return_an_error_message
assert resp.content_type == "application/json"
assert await resp.json() == [
{
"in": "headers",
"loc": ["signature_expired"],
"msg": "invalid datetime format",
"type": "value_error.datetime",
@@ -100,37 +85,3 @@ async def test_get_article_with_valid_header_containing_hyphen_should_be_returne
assert resp.status == 200
assert resp.content_type == "application/json"
assert await resp.json() == {"signature": "2020-10-04T18:01:00"}
async def test_wrong_value_to_header_defined_with_str_enum(aiohttp_client, loop):
app = web.Application()
app.router.add_view("/coord", ViewWithEnumType)
client = await aiohttp_client(app)
resp = await client.get("/coord", headers={"format": "WGS84"})
assert (
await resp.json()
== [
{
"ctx": {"enum_values": ["UMT", "MGRS"]},
"in": "headers",
"loc": ["format"],
"msg": "value is not a valid enumeration member; permitted: 'UMT', 'MGRS'",
"type": "type_error.enum",
}
]
!= {"signature": "2020-10-04T18:01:00"}
)
assert resp.status == 400
assert resp.content_type == "application/json"
async def test_correct_value_to_header_defined_with_str_enum(aiohttp_client, loop):
app = web.Application()
app.router.add_view("/coord", ViewWithEnumType)
client = await aiohttp_client(app)
resp = await client.get("/coord", headers={"format": "UMT"})
assert await resp.json() == {"format": "UMT"}
assert resp.status == 200
assert resp.content_type == "application/json"

View File

@@ -1,5 +1,4 @@
from aiohttp import web
from aiohttp_pydantic import PydanticView
@@ -8,7 +7,7 @@ class ArticleView(PydanticView):
return web.json_response({"path": [author_id, tag, date]})
async def test_get_article_with_correct_path_parameters_should_return_parameters_in_path(
async def test_get_article_without_required_qs_should_return_an_error_message(
aiohttp_client, loop
):
app = web.Application()
@@ -19,23 +18,3 @@ async def test_get_article_with_correct_path_parameters_should_return_parameters
assert resp.status == 200
assert resp.content_type == "application/json"
assert await resp.json() == {"path": ["1234", "music", 1980]}
async def test_get_article_with_wrong_path_parameters_should_return_error(
aiohttp_client, loop
):
app = web.Application()
app.router.add_view("/article/{author_id}/tag/{tag}/before/{date}", ArticleView)
client = await aiohttp_client(app)
resp = await client.get("/article/1234/tag/music/before/now")
assert resp.status == 400
assert resp.content_type == "application/json"
assert await resp.json() == [
{
"in": "path",
"loc": ["date"],
"msg": "value is not a valid integer",
"type": "type_error.integer",
}
]

View File

@@ -1,17 +1,10 @@
from typing import Optional
from aiohttp import web
from aiohttp_pydantic import PydanticView
class ArticleView(PydanticView):
async def get(
self, with_comments: bool, age: Optional[int] = None, nb_items: int = 7
):
return web.json_response(
{"with_comments": with_comments, "age": age, "nb_items": nb_items}
)
async def get(self, with_comments: bool):
return web.json_response({"with_comments": with_comments})
async def test_get_article_without_required_qs_should_return_an_error_message(
@@ -26,7 +19,6 @@ async def test_get_article_without_required_qs_should_return_an_error_message(
assert resp.content_type == "application/json"
assert await resp.json() == [
{
"in": "query string",
"loc": ["with_comments"],
"msg": "field required",
"type": "value_error.missing",
@@ -46,7 +38,6 @@ async def test_get_article_with_wrong_qs_type_should_return_an_error_message(
assert resp.content_type == "application/json"
assert await resp.json() == [
{
"in": "query string",
"loc": ["with_comments"],
"msg": "value could not be parsed to a boolean",
"type": "type_error.bool",
@@ -61,22 +52,7 @@ async def test_get_article_with_valid_qs_should_return_the_parsed_type(
app.router.add_view("/article", ArticleView)
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, "nb_items": 7}
async def test_get_article_with_valid_qs_and_omitted_optional_should_return_default_value(
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"})
assert await resp.json() == {"with_comments": True, "age": None, "nb_items": 7}
assert resp.status == 200
assert resp.content_type == "application/json"
assert await resp.json() == {"with_comments": True}