Create aiohttp_pydantic

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+.pytest_cache

aiohttp_pydantic/__init__.py

@@ -0,0 +1,87 @@
+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 BaseModel, ValidationError
+from typing import Generator, Any
+from aiohttp.web import json_response
+
+
+class PydanticView(AbstractView):
+
+    async def _iter(self) -> StreamResponse:
+        method = getattr(self, self.request.method.lower(), None)
+        resp = await method()
+        return resp
+
+    def __await__(self) -> Generator[Any, None, StreamResponse]:
+        return self._iter().__await__()
+
+    def __init_subclass__(cls, **kwargs):
+        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)
+
+        if 'GET' in allowed_methods:
+            cls.get = inject_qs(cls.get)
+        if 'POST' in allowed_methods:
+            cls.post = inject_body(cls.post)
+        if 'PUT' in allowed_methods:
+            cls.put = inject_body(cls.put)
+
+        for meth_name in METH_ALL:
+            if meth_name not in allowed_methods:
+                setattr(cls, meth_name.lower(), raise_not_allowed)
+
+
+def inject_qs(handler):
+    """
+    Decorator to unpack the query string in the parameters of the web handler
+    regarding annotations.
+    """
+    qs_model_class = type(
+        'QSModel', (BaseModel,),
+        {'__annotations__': handler.__annotations__})
+
+    async def wrapped_handler(self):
+        try:
+            qs = qs_model_class(**self.request.query)
+        except ValidationError as error:
+            return json_response(text=error.json(), status=400)
+            # raise HTTPBadRequest(
+            #     reason='\n'.join(
+            #         f'Error with query string parameter {", ".join(err["loc"])}:'
+            #         f' {err["msg"]}' for err in error.errors()))
+
+        return await handler(self, **qs.dict())
+
+    return wrapped_handler
+
+
+def inject_body(handler):
+    """
+    Decorator to inject the request body as parameter of the web handler
+    regarding annotations.
+    """
+
+    arg_name, model_class = next(
+        ((arg_name, arg_type)
+         for arg_name, arg_type in handler.__annotations__.items()
+         if issubclass(arg_type, BaseModel)), (None, None))
+
+    if arg_name is None:
+        return handler
+
+    async def wrapped_handler(self):
+        body = await self.request.json()
+        try:
+            model = model_class(**body)
+        except ValidationError as error:
+            return json_response(text=error.json(), status=400)
+
+        return await handler(self, **{arg_name: model})
+
+    return wrapped_handler

setup.py

@@ -0,0 +1,30 @@
+from setuptools import setup
+
+
+setup(
+    name='aiohttp_pydantic',
+    version='0.0.1',
+    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.6',
+        'Programming Language :: Python :: 3.7',
+        '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']
+)

tests/test_validation_body.py

@@ -0,0 +1,52 @@
+from pydantic import BaseModel
+from typing import Optional
+from aiohttp import web
+from aiohttp_pydantic import PydanticView
+
+
+class ArticleModel(BaseModel):
+    name: str
+    nb_page: Optional[int]
+
+
+class ArticleView(PydanticView):
+
+    async def post(self, article: ArticleModel):
+        return web.Response(article.dict())
+
+
+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)
+
+    client = await aiohttp_client(app)
+    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'}]
+
+
+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)
+
+    client = await aiohttp_client(app)
+    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'}]
+
+
+async def test_post_a_valid_article_should_return_the_parsed_type(aiohttp_client, loop):
+    app = web.Application()
+    app.router.add_view('/article', ArticleView)
+
+    client = await aiohttp_client(app)
+    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}

tests/test_validation_query_string.py

@@ -0,0 +1,45 @@
+from aiohttp import web
+from aiohttp_pydantic import PydanticView
+
+
+class ArticleView(PydanticView):
+
+    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(aiohttp_client, loop):
+    app = web.Application()
+    app.router.add_view('/article', ArticleView)
+
+    client = await aiohttp_client(app)
+    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'}]
+
+
+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)
+
+    client = await aiohttp_client(app)
+    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'}]
+
+
+async def test_get_article_with_valide_qs_should_return_the_parsed_type(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 resp.status == 200
+    assert resp.content_type == 'application/json'
+    assert await resp.json() == {'with_comments': True}