diff --git a/.travis.yml b/.travis.yml
index 5d7aff7..3f2152e 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -4,7 +4,9 @@ python:
script:
- pytest tests/
install:
+- pip install -U setuptools wheel pip
- pip install -r test_requirements.txt
+- pip install .
deploy:
provider: pypi
username: __token__
diff --git a/README.rst b/README.rst
index 92de6c8..815665d 100644
--- a/README.rst
+++ b/README.rst
@@ -144,3 +144,58 @@ To declare a HTTP headers parameters, you must declare your argument as a `keywo
.. _positional-only parameters: https://www.python.org/dev/peps/pep-0570/
.. _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
+--------------------------------------------
+
+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
+
+.. code-block:: python3
+
+ from aiohttp import web
+ from aiohttp_pydantic import oas
+
+
+ app = web.Application()
+ oas.setup(app)
+
+By default, the route to display the Open Api Specification is /oas but you can change it using
+*url_prefix* parameter
+
+
+.. code-block:: python3
+
+ oas.setup(app, url_prefix='/spec-api')
+
+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
+
+ from aiohttp import web
+ from aiohttp_pydantic import oas
+
+ app = web.Application()
+ sub_app_1 = web.Application()
+
+ oas.setup(app, apps_to_expose=[app, sub_app_1])
+
+Demo
+====
+
+Have a look at `demo`_ for a complete example
+
+.. code-block:: bash
+
+ git clone https://github.com/Maillol/aiohttp-pydantic.git
+ cd aiohttp-pydantic
+ pip install .
+ python -m demo
+
+Go to http://127.0.0.1:8080/oas
+
+
+
+.. _demo: https://github.com/Maillol/aiohttp-pydantic/tree/main/demo
diff --git a/aiohttp_pydantic/__init__.py b/aiohttp_pydantic/__init__.py
index ba6ea72..24f3a94 100644
--- a/aiohttp_pydantic/__init__.py
+++ b/aiohttp_pydantic/__init__.py
@@ -1,3 +1,5 @@
from .view import PydanticView
-__all__ = ("PydanticView",)
+__version__ = "1.0.0"
+
+__all__ = ("PydanticView", "__version__")
diff --git a/aiohttp_pydantic/injectors.py b/aiohttp_pydantic/injectors.py
index 311700e..f96fccc 100644
--- a/aiohttp_pydantic/injectors.py
+++ b/aiohttp_pydantic/injectors.py
@@ -94,6 +94,9 @@ def _parse_func_signature(func: Callable) -> Tuple[dict, dict, dict, dict]:
if param_name == "self":
continue
+ if param_spec.annotation == param_spec.empty:
+ raise RuntimeError(f"The parameter {param_name} must have an annotation")
+
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:
diff --git a/aiohttp_pydantic/oas/__init__.py b/aiohttp_pydantic/oas/__init__.py
new file mode 100644
index 0000000..939ec13
--- /dev/null
+++ b/aiohttp_pydantic/oas/__init__.py
@@ -0,0 +1,26 @@
+from typing import Iterable
+from importlib import resources
+
+import jinja2
+from aiohttp import web
+from .view import get_oas, oas_ui
+from swagger_ui_bundle import swagger_ui_path
+
+
+def setup(
+ app: web.Application,
+ apps_to_expose: Iterable[web.Application] = (),
+ url_prefix: str = "/oas",
+ enable: bool = True,
+):
+ if enable:
+ oas_app = web.Application()
+ oas_app["apps to expose"] = tuple(apps_to_expose) or (app,)
+ oas_app["index template"] = jinja2.Template(
+ resources.read_text("aiohttp_pydantic.oas", "index.j2")
+ )
+ oas_app.router.add_get("/spec", get_oas, name="spec")
+ oas_app.router.add_static("/static", swagger_ui_path, name="static")
+ oas_app.router.add_get("", oas_ui, name="index")
+
+ app.add_subapp(url_prefix, oas_app)
diff --git a/aiohttp_pydantic/oas/index.j2 b/aiohttp_pydantic/oas/index.j2
new file mode 100644
index 0000000..cfbaf02
--- /dev/null
+++ b/aiohttp_pydantic/oas/index.j2
@@ -0,0 +1,69 @@
+{# This updated file is part of swagger_ui_bundle (https://github.com/dtkav/swagger_ui_bundle) #}
+
+
+
+
+ {{ title | default('Swagger UI') }}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/aiohttp_pydantic/oas/struct.py b/aiohttp_pydantic/oas/struct.py
new file mode 100644
index 0000000..4d62a19
--- /dev/null
+++ b/aiohttp_pydantic/oas/struct.py
@@ -0,0 +1,246 @@
+class Info:
+ def __init__(self, spec: dict):
+ self._spec = spec.setdefault("info", {})
+
+ @property
+ def title(self):
+ return self._spec["title"]
+
+ @title.setter
+ def title(self, title):
+ self._spec["title"] = title
+
+ @property
+ def description(self):
+ return self._spec["description"]
+
+ @description.setter
+ def description(self, description):
+ self._spec["description"] = description
+
+ @property
+ def version(self):
+ return self._spec["version"]
+
+ @version.setter
+ def version(self, version):
+ self._spec["version"] = version
+
+
+class RequestBody:
+ def __init__(self, spec: dict):
+ self._spec = spec.setdefault("requestBody", {})
+
+ @property
+ def description(self):
+ return self._spec["description"]
+
+ @description.setter
+ def description(self, description: str):
+ self._spec["description"] = description
+
+ @property
+ def required(self):
+ return self._spec["required"]
+
+ @required.setter
+ def required(self, required: bool):
+ self._spec["required"] = required
+
+ @property
+ def content(self):
+ return self._spec["content"]
+
+ @content.setter
+ def content(self, content: dict):
+ self._spec["content"] = content
+
+
+class Parameter:
+ def __init__(self, spec: dict):
+ self._spec = spec
+
+ @property
+ def name(self) -> str:
+ return self._spec["name"]
+
+ @name.setter
+ def name(self, name: str):
+ self._spec["name"] = name
+
+ @property
+ def in_(self) -> str:
+ return self._spec["in"]
+
+ @in_.setter
+ def in_(self, in_: str):
+ self._spec["in"] = in_
+
+ @property
+ def description(self) -> str:
+ return self._spec["description"]
+
+ @description.setter
+ def description(self, description: str):
+ self._spec["description"] = description
+
+ @property
+ def required(self) -> bool:
+ return self._spec["required"]
+
+ @required.setter
+ def required(self, required: bool):
+ self._spec["required"] = required
+
+ @property
+ def schema(self) -> dict:
+ return self._spec["schema"]
+
+ @schema.setter
+ def schema(self, schema: dict):
+ self._spec["schema"] = schema
+
+
+class Parameters:
+ def __init__(self, spec):
+ self._spec = spec
+ self._spec.setdefault("parameters", [])
+
+ def __getitem__(self, item: int) -> Parameter:
+ if item == len(self._spec["parameters"]):
+ spec = {}
+ self._spec["parameters"].append(spec)
+ else:
+ spec = self._spec["parameters"][item]
+ return Parameter(spec)
+
+
+class OperationObject:
+ def __init__(self, spec: dict):
+ self._spec = spec
+
+ @property
+ def summary(self) -> str:
+ return self._spec["summary"]
+
+ @summary.setter
+ def summary(self, summary: str):
+ self._spec["summary"] = summary
+
+ @property
+ def description(self) -> str:
+ return self._spec["description"]
+
+ @description.setter
+ def description(self, description: str):
+ self._spec["description"] = description
+
+ @property
+ def request_body(self) -> RequestBody:
+ return RequestBody(self._spec)
+
+ @property
+ def parameters(self) -> Parameters:
+ return Parameters(self._spec)
+
+
+class PathItem:
+ def __init__(self, spec: dict):
+ self._spec = spec
+
+ @property
+ def get(self) -> OperationObject:
+ return OperationObject(self._spec.setdefault("get", {}))
+
+ @property
+ def put(self) -> OperationObject:
+ return OperationObject(self._spec.setdefault("put", {}))
+
+ @property
+ def post(self) -> OperationObject:
+ return OperationObject(self._spec.setdefault("post", {}))
+
+ @property
+ def delete(self) -> OperationObject:
+ return OperationObject(self._spec.setdefault("delete", {}))
+
+ @property
+ def options(self) -> OperationObject:
+ return OperationObject(self._spec.setdefault("options", {}))
+
+ @property
+ def head(self) -> OperationObject:
+ return OperationObject(self._spec.setdefault("head", {}))
+
+ @property
+ def patch(self) -> OperationObject:
+ return OperationObject(self._spec.setdefault("patch", {}))
+
+ @property
+ def trace(self) -> OperationObject:
+ return OperationObject(self._spec.setdefault("trace", {}))
+
+
+class Paths:
+ def __init__(self, spec: dict):
+ self._spec = spec.setdefault("paths", {})
+
+ def __getitem__(self, path: str) -> PathItem:
+ spec = self._spec.setdefault(path, {})
+ return PathItem(spec)
+
+
+class Server:
+ def __init__(self, spec: dict):
+ self._spec = spec
+
+ @property
+ def url(self) -> str:
+ return self._spec["url"]
+
+ @url.setter
+ def url(self, url: str):
+ self._spec["url"] = url
+
+ @property
+ def description(self) -> str:
+ return self._spec["url"]
+
+ @description.setter
+ def description(self, description: str):
+ self._spec["description"] = description
+
+
+class Servers:
+ def __init__(self, spec: dict):
+ self._spec = spec
+ self._spec.setdefault("servers", [])
+
+ def __getitem__(self, item: int) -> Server:
+ if item == len(self._spec["servers"]):
+ spec = {}
+ self._spec["servers"].append(spec)
+ else:
+ spec = self._spec["servers"][item]
+ return Server(spec)
+
+
+class OpenApiSpec3:
+ def __init__(self):
+ self._spec = {"openapi": "3.0.0"}
+
+ @property
+ def info(self) -> Info:
+ return Info(self._spec)
+
+ @property
+ def servers(self) -> Servers:
+ return Servers(self._spec)
+
+ @property
+ def paths(self) -> Paths:
+ return Paths(self._spec)
+
+ @property
+ def spec(self):
+ return self._spec
diff --git a/aiohttp_pydantic/oas/view.py b/aiohttp_pydantic/oas/view.py
new file mode 100644
index 0000000..52e0408
--- /dev/null
+++ b/aiohttp_pydantic/oas/view.py
@@ -0,0 +1,85 @@
+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
+
+
+JSON_SCHEMA_TYPES = {float: "number", str: "string", int: "integer"}
+
+
+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:
+ mtd.request_body.content = {
+ "application/json": {"schema": next(iter(body_args.values())).schema()}
+ }
+
+ 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_]}
+
+ 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_]}
+
+ 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):
+ """
+ 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():
+ 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 to serve the swagger-ui to read open api specification of application.
+ """
+ template = request.app["index template"]
+
+ static_url = request.app.router["static"].url_for(filename="")
+ spec_url = request.app.router["spec"].url_for()
+ host = request.url.origin()
+
+ return Response(
+ text=template.render(
+ {
+ "openapi_spec_url": host.with_path(str(spec_url)),
+ "static_url": host.with_path(str(static_url)),
+ }
+ ),
+ content_type="text/html",
+ charset="utf-8",
+ )
diff --git a/aiohttp_pydantic/view.py b/aiohttp_pydantic/view.py
index 05fdc41..e5b8587 100644
--- a/aiohttp_pydantic/view.py
+++ b/aiohttp_pydantic/view.py
@@ -1,11 +1,10 @@
from inspect import iscoroutinefunction
-
from aiohttp.abc import AbstractView
from aiohttp.hdrs import METH_ALL
from aiohttp.web_exceptions import HTTPMethodNotAllowed
from aiohttp.web_response import StreamResponse
from pydantic import ValidationError
-from typing import Generator, Any, Callable, List, Iterable
+from typing import Generator, Any, Callable, Type, Iterable
from aiohttp.web import json_response
from functools import update_wrapper
@@ -34,21 +33,21 @@ class PydanticView(AbstractView):
return self._iter().__await__()
def __init_subclass__(cls, **kwargs):
- allowed_methods = {
+ cls.allowed_methods = {
meth_name for meth_name in METH_ALL if hasattr(cls, meth_name.lower())
}
- async def raise_not_allowed(self):
- raise HTTPMethodNotAllowed(self.request.method, allowed_methods)
-
for meth_name in METH_ALL:
- if meth_name not in allowed_methods:
- setattr(cls, meth_name.lower(), raise_not_allowed)
+ if meth_name not in cls.allowed_methods:
+ setattr(cls, meth_name.lower(), cls.raise_not_allowed)
else:
handler = getattr(cls, meth_name.lower())
decorated_handler = inject_params(handler, cls.parse_func_signature)
setattr(cls, meth_name.lower(), decorated_handler)
+ async def raise_not_allowed(self):
+ raise HTTPMethodNotAllowed(self.request.method, self.allowed_methods)
+
@staticmethod
def parse_func_signature(func: Callable) -> Iterable[AbstractInjector]:
path_args, body_args, qs_args, header_args = _parse_func_signature(func)
@@ -90,3 +89,13 @@ def inject_params(
update_wrapper(wrapped_handler, handler)
return wrapped_handler
+
+
+def is_pydantic_view(obj) -> bool:
+ """
+ Return True if obj is a PydanticView subclass else False.
+ """
+ try:
+ return issubclass(obj, PydanticView)
+ except TypeError:
+ return False
diff --git a/demo/__init__.py b/demo/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/demo/__main__.py b/demo/__main__.py
new file mode 100644
index 0000000..58025fb
--- /dev/null
+++ b/demo/__main__.py
@@ -0,0 +1,25 @@
+from aiohttp import web
+
+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)
diff --git a/demo/model.py b/demo/model.py
new file mode 100644
index 0000000..3e1d4c7
--- /dev/null
+++ b/demo/model.py
@@ -0,0 +1,43 @@
+from pydantic import BaseModel
+
+
+class Pet(BaseModel):
+ id: int
+ name: str
+
+
+class Model:
+ """
+ To keep simple this demo, we use a simple dict as database to
+ store the models.
+ """
+
+ class NotFound(KeyError):
+ """
+ Raised when a pet is not found.
+ """
+
+ def __init__(self):
+ self.storage = {}
+
+ def add_pet(self, pet: Pet):
+ self.storage[pet.id] = pet
+
+ def remove_pet(self, id: int):
+ try:
+ del self.storage[id]
+ except KeyError as error:
+ raise self.NotFound(str(error))
+
+ def update_pet(self, id: int, pet: Pet):
+ self.remove_pet(id)
+ self.add_pet(pet)
+
+ def find_pet(self, id: int):
+ try:
+ return self.storage[id]
+ except KeyError as error:
+ raise self.NotFound(str(error))
+
+ def list_pets(self):
+ return list(self.storage.values())
diff --git a/demo/view.py b/demo/view.py
new file mode 100644
index 0000000..fe12ffc
--- /dev/null
+++ b/demo/view.py
@@ -0,0 +1,28 @@
+from aiohttp_pydantic import PydanticView
+from aiohttp import web
+
+from .model import Pet
+
+
+class PetCollectionView(PydanticView):
+ 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):
+ self.request.app["model"].add_pet(pet)
+ return web.json_response(pet.dict())
+
+
+class PetItemView(PydanticView):
+ 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):
+ self.request.app["model"].update_pet(id, pet)
+ return web.json_response(pet.dict())
+
+ async def delete(self, id: int, /):
+ self.request.app["model"].remove_pet(id)
+ return web.json_response(id)
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..6228f7d
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,6 @@
+[build-system]
+requires = [
+ "setuptools >= 46.4.0",
+ "wheel",
+]
+build-backend = "setuptools.build_meta"
\ No newline at end of file
diff --git a/setup.cfg b/setup.cfg
new file mode 100644
index 0000000..317900c
--- /dev/null
+++ b/setup.cfg
@@ -0,0 +1,47 @@
+[metadata]
+name = aiohttp_pydantic
+version = attr: aiohttp_pydantic.__version__
+url = https://github.com/Maillol/aiohttp-pydantic
+author = Vincent Maillol
+author_email = vincent.maillol@gmail.com
+description = Aiohttp View using pydantic to validate request body and query sting regarding method annotations.
+long_description = file: README.rst
+keywords =
+ aiohttp
+ pydantic
+ annotations
+ validation
+license = MIT
+classifiers =
+ Intended Audience :: Developers
+ Intended Audience :: Information Technology
+ Programming Language :: Python
+ Programming Language :: Python :: 3
+ Programming Language :: Python :: 3 :: Only
+ Programming Language :: Python :: 3.8
+ Programming Language :: Python :: 3.9
+ Topic :: Software Development :: Libraries :: Application Frameworks
+ Framework :: AsyncIO
+ License :: OSI Approved :: MIT License
+
+[options]
+zip_safe = False
+include_package_data = True
+packages = find:
+python_requires = >=3.8
+install_requires =
+ aiohttp
+ pydantic
+ swagger-ui-bundle
+
+[options.extras_require]
+test = pytest; pytest-aiohttp
+
+
+[options.packages.find]
+exclude =
+ tests
+ demo
+
+[options.package_data]
+aiohttp_pydantic.oas = index.j2
diff --git a/setup.py b/setup.py
index 06ddc7a..6068493 100644
--- a/setup.py
+++ b/setup.py
@@ -1,28 +1,3 @@
from setuptools import setup
-
-setup(
- name='aiohttp_pydantic',
- version='1.0.0',
- description='Aiohttp View using pydantic to validate request body and query sting regarding method annotation',
- keywords='aiohttp pydantic annotation unpack inject validate',
- author='Vincent Maillol',
- author_email='vincent.maillol@gmail.com',
- url='https://github.com/Maillol/aiohttp-pydantic',
- license='MIT',
- packages=['aiohttp_pydantic'],
- classifiers=[
- 'Intended Audience :: Developers',
- 'Intended Audience :: Information Technology',
- 'Programming Language :: Python',
- 'Programming Language :: Python :: 3',
- 'Programming Language :: Python :: 3 :: Only',
- 'Programming Language :: Python :: 3.8',
- 'Programming Language :: Python :: 3.9',
- 'Topic :: Software Development :: Libraries :: Application Frameworks',
- 'Framework :: AsyncIO',
- 'License :: OSI Approved :: MIT License'
- ],
- python_requires='>=3.6',
- install_requires=['aiohttp', 'pydantic']
-)
+setup()
diff --git a/test_requirements.txt b/test_requirements.txt
index 87f0130..3ac21f8 100644
--- a/test_requirements.txt
+++ b/test_requirements.txt
@@ -1,3 +1,3 @@
-.
pytest==6.1.1
-pytest-aiohttp==0.3.0
\ No newline at end of file
+pytest-aiohttp==0.3.0
+typing_extensions>=3.6.5
\ No newline at end of file
diff --git a/tests/test_oas/__init__.py b/tests/test_oas/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/test_oas/test_view.py b/tests/test_oas/test_view.py
new file mode 100644
index 0000000..78f39ec
--- /dev/null
+++ b/tests/test_oas/test_view.py
@@ -0,0 +1,129 @@
+from pydantic.main import BaseModel
+from aiohttp_pydantic import PydanticView, oas
+from aiohttp import web
+
+import pytest
+
+
+class Pet(BaseModel):
+ id: int
+ name: str
+
+
+class PetCollectionView(PydanticView):
+ async def get(self):
+ return web.json_response()
+
+ async def post(self, pet: Pet):
+ return web.json_response()
+
+
+class PetItemView(PydanticView):
+ 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, /):
+ return web.json_response()
+
+
+@pytest.fixture
+async def generated_oas(aiohttp_client, loop) -> web.Application:
+ app = web.Application()
+ app.router.add_view("/pets", PetCollectionView)
+ app.router.add_view("/pets/{id}", PetItemView)
+ oas.setup(app)
+
+ client = await aiohttp_client(app)
+ response = await client.get("/oas/spec")
+ assert response.status == 200
+ assert response.content_type == "application/json"
+ return await response.json()
+
+
+async def test_generated_oas_should_have_pets_paths(generated_oas):
+ assert "/pets" in generated_oas["paths"]
+
+
+async def test_pets_route_should_have_get_method(generated_oas):
+ assert generated_oas["paths"]["/pets"]["get"] == {}
+
+
+async def test_pets_route_should_have_post_method(generated_oas):
+ assert generated_oas["paths"]["/pets"]["post"] == {
+ "requestBody": {
+ "content": {
+ "application/json": {
+ "schema": {
+ "properties": {
+ "id": {"title": "Id", "type": "integer"},
+ "name": {"title": "Name", "type": "string"},
+ },
+ "required": ["id", "name"],
+ "title": "Pet",
+ "type": "object",
+ }
+ }
+ }
+ }
+ }
+
+
+async def test_generated_oas_should_have_pets_id_paths(generated_oas):
+ assert "/pets/{id}" in generated_oas["paths"]
+
+
+async def test_pets_id_route_should_have_delete_method(generated_oas):
+ assert generated_oas["paths"]["/pets/{id}"]["delete"] == {
+ "parameters": [
+ {
+ "in": "path",
+ "name": "id",
+ "required": True,
+ "schema": {"type": "integer"},
+ }
+ ]
+ }
+
+
+async def test_pets_id_route_should_have_get_method(generated_oas):
+ assert generated_oas["paths"]["/pets/{id}"]["get"] == {
+ "parameters": [
+ {
+ "in": "path",
+ "name": "id",
+ "required": True,
+ "schema": {"type": "integer"},
+ }
+ ]
+ }
+
+
+async def test_pets_id_route_should_have_put_method(generated_oas):
+ assert generated_oas["paths"]["/pets/{id}"]["put"] == {
+ "parameters": [
+ {
+ "in": "path",
+ "name": "id",
+ "required": True,
+ "schema": {"type": "integer"},
+ }
+ ],
+ "requestBody": {
+ "content": {
+ "application/json": {
+ "schema": {
+ "properties": {
+ "id": {"title": "Id", "type": "integer"},
+ "name": {"title": "Name", "type": "string"},
+ },
+ "required": ["id", "name"],
+ "title": "Pet",
+ "type": "object",
+ }
+ }
+ }
+ },
+ }
diff --git a/tests/test_parse_func_signature.py b/tests/test_parse_func_signature.py
index 6170df5..3c91e9f 100644
--- a/tests/test_parse_func_signature.py
+++ b/tests/test_parse_func_signature.py
@@ -9,7 +9,6 @@ class User(BaseModel):
def test_parse_func_signature():
-
def body_only(self, user: User):
pass
@@ -37,13 +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(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})
-
+ 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},
+ )
diff --git a/tests/test_validation_body.py b/tests/test_validation_body.py
index 9a7b14b..33951fb 100644
--- a/tests/test_validation_body.py
+++ b/tests/test_validation_body.py
@@ -10,43 +10,50 @@ class ArticleModel(BaseModel):
class ArticleView(PydanticView):
-
async def post(self, article: ArticleModel):
return web.json_response(article.dict())
-async def test_post_an_article_without_required_field_should_return_an_error_message(aiohttp_client, loop):
+async def test_post_an_article_without_required_field_should_return_an_error_message(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
- resp = await client.post('/article', json={})
+ resp = await client.post("/article", json={})
assert resp.status == 400
- assert resp.content_type == 'application/json'
- assert await resp.json() == [{'loc': ['name'],
- 'msg': 'field required',
- 'type': 'value_error.missing'}]
+ assert resp.content_type == "application/json"
+ assert await resp.json() == [
+ {"loc": ["name"], "msg": "field required", "type": "value_error.missing"}
+ ]
-async def test_post_an_article_with_wrong_type_field_should_return_an_error_message(aiohttp_client, loop):
+async def test_post_an_article_with_wrong_type_field_should_return_an_error_message(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
- resp = await client.post('/article', json={'name': 'foo', 'nb_page': 'foo'})
+ resp = await client.post("/article", json={"name": "foo", "nb_page": "foo"})
assert resp.status == 400
- assert resp.content_type == 'application/json'
- assert await resp.json() == [{'loc': ['nb_page'],
- 'msg': 'value is not a valid integer',
- 'type': 'type_error.integer'}]
+ assert resp.content_type == "application/json"
+ assert await resp.json() == [
+ {
+ "loc": ["nb_page"],
+ "msg": "value is not a valid integer",
+ "type": "type_error.integer",
+ }
+ ]
async def test_post_a_valid_article_should_return_the_parsed_type(aiohttp_client, loop):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
- resp = await client.post('/article', json={'name': 'foo', 'nb_page': 3})
+ resp = await client.post("/article", json={"name": "foo", "nb_page": 3})
assert resp.status == 200
- assert resp.content_type == 'application/json'
- assert await resp.json() == {'name': 'foo', 'nb_page': 3}
+ assert resp.content_type == "application/json"
+ assert await resp.json() == {"name": "foo", "nb_page": 3}
diff --git a/tests/test_validation_header.py b/tests/test_validation_header.py
index c198fb1..511f45c 100644
--- a/tests/test_validation_header.py
+++ b/tests/test_validation_header.py
@@ -5,7 +5,6 @@ import json
class JSONEncoder(json.JSONEncoder):
-
def default(self, o):
if isinstance(o, datetime):
return o.isoformat()
@@ -14,54 +13,75 @@ class JSONEncoder(json.JSONEncoder):
class ArticleView(PydanticView):
-
async def get(self, *, signature_expired: datetime):
- return web.json_response({'signature': signature_expired}, dumps=JSONEncoder().encode)
+ return web.json_response(
+ {"signature": signature_expired}, dumps=JSONEncoder().encode
+ )
-async def test_get_article_without_required_header_should_return_an_error_message(aiohttp_client, loop):
+async def test_get_article_without_required_header_should_return_an_error_message(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
- resp = await client.get('/article', headers={})
+ resp = await client.get("/article", headers={})
assert resp.status == 400
- assert resp.content_type == 'application/json'
- assert await resp.json() == [{'loc': ['signature_expired'],
- 'msg': 'field required',
- 'type': 'value_error.missing'}]
+ assert resp.content_type == "application/json"
+ assert await resp.json() == [
+ {
+ "loc": ["signature_expired"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
-async def test_get_article_with_wrong_header_type_should_return_an_error_message(aiohttp_client, loop):
+async def test_get_article_with_wrong_header_type_should_return_an_error_message(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
- resp = await client.get('/article', headers={'signature_expired': 'foo'})
+ resp = await client.get("/article", headers={"signature_expired": "foo"})
assert resp.status == 400
- assert resp.content_type == 'application/json'
- assert await resp.json() == [{'loc': ['signature_expired'],
- 'msg': 'invalid datetime format',
- 'type': 'value_error.datetime'}]
+ assert resp.content_type == "application/json"
+ assert await resp.json() == [
+ {
+ "loc": ["signature_expired"],
+ "msg": "invalid datetime format",
+ "type": "value_error.datetime",
+ }
+ ]
-async def test_get_article_with_valid_header_should_return_the_parsed_type(aiohttp_client, loop):
+async def test_get_article_with_valid_header_should_return_the_parsed_type(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
- resp = await client.get('/article', headers={'signature_expired': '2020-10-04T18:01:00'})
+ resp = await client.get(
+ "/article", headers={"signature_expired": "2020-10-04T18:01:00"}
+ )
assert resp.status == 200
- assert resp.content_type == 'application/json'
- assert await resp.json() == {'signature': '2020-10-04T18:01:00'}
+ assert resp.content_type == "application/json"
+ assert await resp.json() == {"signature": "2020-10-04T18:01:00"}
-async def test_get_article_with_valid_header_containing_hyphen_should_be_returned(aiohttp_client, loop):
+async def test_get_article_with_valid_header_containing_hyphen_should_be_returned(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
- resp = await client.get('/article', headers={'Signature-Expired': '2020-10-04T18:01:00'})
+ resp = await client.get(
+ "/article", headers={"Signature-Expired": "2020-10-04T18:01:00"}
+ )
assert resp.status == 200
- assert resp.content_type == 'application/json'
- assert await resp.json() == {'signature': '2020-10-04T18:01:00'}
+ assert resp.content_type == "application/json"
+ assert await resp.json() == {"signature": "2020-10-04T18:01:00"}
diff --git a/tests/test_validation_path.py b/tests/test_validation_path.py
index 832e363..fe7466f 100644
--- a/tests/test_validation_path.py
+++ b/tests/test_validation_path.py
@@ -3,18 +3,18 @@ from aiohttp_pydantic import PydanticView
class ArticleView(PydanticView):
-
- async def get(self, author_id: str, tag: str, date: int, /):
- return web.json_response({'path': [author_id, tag, date]})
+ async def get(self, author_id: str, tag: str, date: int, /):
+ return web.json_response({"path": [author_id, tag, date]})
-async def test_get_article_without_required_qs_should_return_an_error_message(aiohttp_client, loop):
+async def test_get_article_without_required_qs_should_return_an_error_message(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article/{author_id}/tag/{tag}/before/{date}', ArticleView)
+ 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/1980')
+ resp = await client.get("/article/1234/tag/music/before/1980")
assert resp.status == 200
- assert resp.content_type == 'application/json'
- assert await resp.json() == {'path': ['1234', 'music', 1980]}
-
+ assert resp.content_type == "application/json"
+ assert await resp.json() == {"path": ["1234", "music", 1980]}
diff --git a/tests/test_validation_query_string.py b/tests/test_validation_query_string.py
index bd43366..3df2336 100644
--- a/tests/test_validation_query_string.py
+++ b/tests/test_validation_query_string.py
@@ -3,43 +3,56 @@ from aiohttp_pydantic import PydanticView
class ArticleView(PydanticView):
-
async def get(self, with_comments: bool):
- return web.json_response({'with_comments': with_comments})
+ return web.json_response({"with_comments": with_comments})
-async def test_get_article_without_required_qs_should_return_an_error_message(aiohttp_client, loop):
+async def test_get_article_without_required_qs_should_return_an_error_message(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
- resp = await client.get('/article')
+ resp = await client.get("/article")
assert resp.status == 400
- assert resp.content_type == 'application/json'
- assert await resp.json() == [{'loc': ['with_comments'],
- 'msg': 'field required',
- 'type': 'value_error.missing'}]
+ assert resp.content_type == "application/json"
+ assert await resp.json() == [
+ {
+ "loc": ["with_comments"],
+ "msg": "field required",
+ "type": "value_error.missing",
+ }
+ ]
-async def test_get_article_with_wrong_qs_type_should_return_an_error_message(aiohttp_client, loop):
+async def test_get_article_with_wrong_qs_type_should_return_an_error_message(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
- resp = await client.get('/article', params={'with_comments': 'foo'})
+ resp = await client.get("/article", params={"with_comments": "foo"})
assert resp.status == 400
- assert resp.content_type == 'application/json'
- assert await resp.json() == [{'loc': ['with_comments'],
- 'msg': 'value could not be parsed to a boolean',
- 'type': 'type_error.bool'}]
+ assert resp.content_type == "application/json"
+ assert await resp.json() == [
+ {
+ "loc": ["with_comments"],
+ "msg": "value could not be parsed to a boolean",
+ "type": "type_error.bool",
+ }
+ ]
-async def test_get_article_with_valid_qs_should_return_the_parsed_type(aiohttp_client, loop):
+async def test_get_article_with_valid_qs_should_return_the_parsed_type(
+ aiohttp_client, loop
+):
app = web.Application()
- app.router.add_view('/article', ArticleView)
+ 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.content_type == 'application/json'
- assert await resp.json() == {'with_comments': True}
+ assert resp.content_type == "application/json"
+ assert await resp.json() == {"with_comments": True}