Aiohttp pydantic - Aiohttp View to validate and parse request

How to install

$ pip install aiohttp_pydantic

Example:

from typing import Optional

from aiohttp import web
from aiohttp_pydantic import PydanticView
from pydantic import BaseModel

# Use pydantic BaseModel to validate request body
class ArticleModel(BaseModel):
    name: str
    nb_page: Optional[int]


# Create your PydanticView and add annotations.
class ArticleView(PydanticView):

    async def post(self, article: ArticleModel):
        return web.json_response({'name': article.name,
                                  'number_of_page': article.nb_page})

    async def get(self, with_comments: Optional[bool]):
        return web.json_response({'with_comments': with_comments})


app = web.Application()
app.router.add_view('/article', ArticleView)
web.run_app(app)

$ curl -X GET http://127.0.0.1:8080/article?with_comments=a
[
  {
    "loc": [
      "with_comments"
    ],
    "msg": "value could not be parsed to a boolean",
    "type": "type_error.bool"
  }
]

$ curl -X GET http://127.0.0.1:8080/article?with_comments=yes
{"with_comments": true}

$ curl -H "Content-Type: application/json" -X post http://127.0.0.1:8080/article --data '{}'
[
  {
    "loc": [
      "name"
    ],
    "msg": "field required",
    "type": "value_error.missing"
  }
]

$ curl -H "Content-Type: application/json" -X post http://127.0.0.1:8080/article --data '{"name": "toto", "nb_page": "3"}'
{"name": "toto", "number_of_page": 3}

API:

Inject Path Parameters

To declare a path parameters, you must declare your argument as a positional-only parameters:

Example:

class AccountView(PydanticView):
    async def get(self, customer_id: str, account_id: str, /):
        ...

app = web.Application()
app.router.add_get('/customers/{customer_id}/accounts/{account_id}', AccountView)

Inject Query String Parameters

To declare a query parameters, you must declare your argument as simple argument:

class AccountView(PydanticView):
    async def get(self, customer_id: str):
        ...

app = web.Application()
app.router.add_get('/customers', AccountView)

Inject Request Body

To declare a body parameters, you must declare your argument as a simple argument annotated with pydantic Model.

class Customer(BaseModel):
    first_name: str
    last_name: str

class CustomerView(PydanticView):
    async def post(self, customer: Customer):
        ...

app = web.Application()
app.router.add_view('/customers', CustomerView)

Inject HTTP headers

To declare a HTTP headers parameters, you must declare your argument as a keyword-only argument.

class CustomerView(PydanticView):
    async def get(self, *, authorization: str, expire_at: datetime):
        ...

app = web.Application()
app.router.add_view('/customers', CustomerView)

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

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

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

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

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

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

Have a look at demo for a complete example

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

Metadata

Release files for aiohttp-pydantic 1.3.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Built distribution (wheel)

Table of built distributions (wheels) for aiohttp-pydantic 1.3.0
File	Interpreter	ABI	Platform	Reset
aiohttp_pydantic-1.3.0-py3-none-any.whl 13.1 kB Oct 30, 2020	Python 3	none	any	Details

Release files / aiohttp_pydantic-1.3.0-py3-none-any.whl

Download URL	aiohttp_pydantic-1.3.0-py3-none-any.whl
Size	13.1 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`433513e02a9ad86109ee56d587681edda95b7a832c6f689316ca1e0b095fcbd3`
BLAKE2b-256 checksum How to use checksums	`f3222e20cc6ea7fcbb6838d6d54c1d4a5f009d14f02096d680937abd19ee843e`
Upload date	Oct 30, 2020
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/3.2.0 pkginfo/1.6.1 requests/2.24.0 setuptools/50.3.2 requests-toolbelt/0.9.1 tqdm/4.51.0 CPython/3.8.0`