Aiohttp pydantic - Aiohttp View to validate and parse request
How to install
$ pip install aiohttp_pydanticExample:
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 (OAS)
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 specific aiohttp sub-applications. on the same route, you must use apps_to_expose parameter.
from aiohttp import web
from aiohttp_pydantic import oas
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
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 demoGo to http://127.0.0.1:8080/oas
You can generate the OAS in a json file using the command:
python -m aiohttp_pydantic.oas demo.main1 maintainer
Metadata
1 maintainer
Release files for aiohttp-pydantic 1.4.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| aiohttp_pydantic-1.4.0-py3-none-any.whl | Python 3 | none | any | Details |
Release files / aiohttp_pydantic-1.4.0-py3-none-any.whl
| Download URL | aiohttp_pydantic-1.4.0-py3-none-any.whl |
|---|---|
| Size | 15.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
5f8b4d35454bf2d43704f0edb887b350fa4270d84bc0014172c97b36ad959b0b
|
|
BLAKE2b-256 checksum How to use checksums |
a7ab46837ebb10ee97d15e4a0e32869707e9e518d4d80fd2589b58b6936a2f88
|
| Upload date | |
|
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
|
