Aiohttp pydantic - Aiohttp View to validate and parse request
=============================================================
.. image:: https://github.com/Maillol/aiohttp-pydantic/actions/workflows/install-package-and-test.yml/badge.svg
:target: https://github.com/Maillol/aiohttp-pydantic/actions/workflows/install-package-and-test.yml
:alt: CI Status
.. 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
--------------
.. code-block:: bash
$ pip install aiohttp_pydantic
Example:
--------
.. code-block:: python3
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: bool=False):
return web.json_response({'with_comments': with_comments})
app = web.Application()
app.router.add_view('/article', ArticleView)
web.run_app(app)
.. code-block:: bash
$ curl -X GET http://127.0.0.1:8080/article?with_comments=a
[
{
"in": "query string",
"loc": [
"with_comments"
],
"msg": "Input should be a valid boolean, unable to interpret input",
"input": "a",
"type": "bool_parsing"
}
]
$ 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 '{}'
[
{
"in": "body",
"loc": [
"name"
],
"msg": "Field required",
"input": {},
"type": "missing"
},
{
"in": "body",
"loc": [
"nb_page"
],
"msg": "Field required",
"input": {},
"type": "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}
Example using view function handler
-----------------------------------
.. code-block:: python3
from typing import Optional
from aiohttp import web
from aiohttp_pydantic.decorator import inject_params
from pydantic import BaseModel
# Use pydantic BaseModel to validate request body
class ArticleModel(BaseModel):
name: str
nb_page: Optional[int]
# Create your function decorated by 'inject_params' and add annotations.
@inject_params
async def post(article: ArticleModel):
return web.json_response({'name': article.name,
'number_of_page': article.nb_page})
# If you need request
@inject_params.and_request
async def get(request, with_comments: bool = False):
request.app["logger"]("OK")
return web.json_response({'with_comments': with_comments})
app = web.Application()
app["logger"] = print
app.router.add_post('/article', post)
app.router.add_get('/article', get)
web.run_app(app)
API:
----
Inject Path Parameters
~~~~~~~~~~~~~~~~~~~~~~
To declare a path parameter, you must declare your argument as a `positional-only parameters`_:
Example:
.. code-block:: python3
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 parameter, you must declare your argument as a simple argument:
.. code-block:: python3
class AccountView(PydanticView):
async def get(self, customer_id: Optional[str] = None):
...
app = web.Application()
app.router.add_get('/customers', AccountView)
A query string parameter is generally optional and we do not want to force the user to set it in the URL.
It's recommended to define a default value. It's possible to get a multiple value for the same parameter using
the List type
.. code-block:: python3
from typing import List
from pydantic import Field
class AccountView(PydanticView):
async def get(self, tags: List[str] = Field(default_factory=list)):
...
app = web.Application()
app.router.add_get('/customers', AccountView)
Inject Request Body
~~~~~~~~~~~~~~~~~~~
To declare a body parameter, you must declare your argument as a simple argument annotated with `pydantic Model`_.
.. code-block:: python3
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 parameter, you must declare your argument as a `keyword-only argument`_.
.. code-block:: python3
class CustomerView(PydanticView):
async def get(self, *, authorization: str, expire_at: datetime):
...
app = web.Application()
app.router.add_view('/customers', CustomerView)
.. _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/
File Upload
-----------
You can receive files in addition to Pydantic data in your views. Here’s an example of how to use it:
Usage Example
Suppose you want to create an API that accepts a book (with a title and a number of pages) as well as two files
representing the pages of the book. Here’s how you can do it:
.. code-block:: python3
from aiohttp import web
from aiohttp_pydantic import PydanticView
from aiohttp_pydantic.uploaded_file import UploadedFile
from pydantic import BaseModel
class BookModel(BaseModel):
title: str
nb_page: int
class BookAndUploadFileView(PydanticView):
async def post(self, book: BookModel, page_1: UploadedFile, page_2: UploadedFile):
content_1 = (await page_1.read()).decode("utf-8")
content_2 = (await page_2.read()).decode("utf-8")
return web.json_response(
{"book": book.model_dump(), "content_1": content_1, "content_2": content_2},
status=201,
)
Implementation Details
~~~~~~~~~~~~~~~~~~~~~~
Files are represented by instances of UploadedFile, which wrap an `aiohttp.BodyPartReader`_.
UploadedFile exposes the read() and read_chunk() methods, allowing you to read the content of uploaded files asynchronously. You can use read() to get the complete content or read_chunk() to read chunks of data at a time.
Constraints to Consider
~~~~~~~~~~~~~~~~~~~~~~~
1 - Argument Order: If you use both Pydantic models and UploadedFile, you must always define BaseModel
type arguments before UploadedFile type arguments. This ensures proper processing of the data.
2 - File Reading Order: UploadedFile instances must be read in the order they are declared in the method.
Since files are not pre-loaded in memory or on disk, it is important to respect this order.
If the reading order is not respected, a MultipartReadingError is raised.
.. _aiohttp.BodyPartReader: https://docs.aiohttp.org/en/stable/multipart_reference.html#aiohttp.BodyPartReader
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
.. 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 specific aiohttp sub-applications.
on the same route, you must use *apps_to_expose* parameter.
.. code-block:: python3
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])
You can change the title or the version of the generated open api specification using
*title_spec* and *version_spec* parameters:
.. code-block:: python3
oas.setup(app, title_spec="My application", version_spec="1.2.3")
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
The docstring of methods will be parsed to fill the descriptions in the
Open Api Specification.
.. 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]]:
"""
Find all pets
Tags: 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]:
"""
Add a new pet to the store
Tags: pet
Status Codes:
201: The pet is created
"""
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]]:
"""
Find a pet by ID
Tags: pet
Status Codes:
200: Successful operation
404: Pet not found
"""
pet = self.request.app["model"].find_pet(id)
return web.json_response(pet.dict())
async def put(self, id: int, /, pet: Pet) -> r200[Pet]:
"""
Update an existing pet
Tags: pet
Status Codes:
200: successful operation
"""
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)
Group parameters
----------------
If your method has lot of parameters you can group them together inside one or several Groups.
.. code-block:: python3
from aiohttp_pydantic.injectors import Group
class Pagination(Group):
page_num: int = 1
page_size: int = 15
class ArticleView(PydanticView):
async def get(self, page: Pagination):
articles = Article.get(page.page_num, page.page_size)
...
The parameters page_num and page_size are expected in the query string, and
set inside a Pagination object passed as page parameter.
The code above is equivalent to:
.. code-block:: python3
class ArticleView(PydanticView):
async def get(self, page_num: int = 1, page_size: int = 15):
articles = Article.get(page_num, page_size)
...
You can add methods or properties to your Group.
.. code-block:: python3
class Pagination(Group):
page_num: int = 1
page_size: int = 15
@property
def num(self):
return self.page_num
@property
def size(self):
return self.page_size
def slice(self):
return slice(self.num, self.size)
class ArticleView(PydanticView):
async def get(self, page: Pagination):
articles = Article.get(page.num, page.size)
...
Custom Validation error
-----------------------
You can redefine the on_validation_error hook in your PydanticView
.. code-block:: python3
class PetView(PydanticView):
async def on_validation_error(self,
exception: ValidationError,
context: str):
errors = exception.errors()
for error in errors:
error["in"] = context # context is "body", "headers", "path" or "query string"
error["custom"] = "your custom field ..."
return json_response(data=errors, status=400)
If you use function based view:
.. code-block:: python3
async def custom_error(exception: ValidationError,
context: str):
errors = exception.errors()
for error in errors:
error["in"] = context # context is "body", "headers", "path" or "query string"
error["custom"] = "your custom field ..."
return json_response(data=errors, status=400)
@inject_params(on_validation_error=custom_error)
async def get(with_comments: bool = False):
...
@inject_params.and_request(on_validation_error=custom_error)
async def get(request, with_comments: bool = False):
...
A tip to use the same error handling on each view
.. code-block:: python3
inject_params = inject_params(on_validation_error=custom_error)
@inject_params
async def post(article: ArticleModel):
return web.json_response({'name': article.name,
'number_of_page': article.nb_page})
@inject_params.and_request
async def get(request, with_comments: bool = False):
return web.json_response({'with_comments': with_comments})
Add security to the endpoints
-----------------------------
aiohttp_pydantic provides a basic way to add security to the endpoints. You can define the security
on the setup level using the *security* parameter and then mark view methods that will require this security schema.
.. code-block:: python3
from aiohttp import web
from aiohttp_pydantic import oas
app = web.Application()
oas.setup(app, security={"APIKeyHeader": {"type": "apiKey", "in": "header", "name": "Authorization"}})
And then mark the view method with the *security* descriptor
.. 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]]:
"""
Find all pets
Security: APIKeyHeader
Tags: 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]:
"""
Add a new pet to the store
Tags: pet
Status Codes:
201: The pet is created
"""
self.request.app["model"].add_pet(pet)
return web.json_response(pet.dict())
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
You can generate the OAS in a json or yaml file using the aiohttp_pydantic.oas command:
.. code-block:: bash
python -m aiohttp_pydantic.oas demo.main
.. code-block:: bash
$ python3 -m aiohttp_pydantic.oas --help
usage: __main__.py [-h] [-b FILE] [-o FILE] [-f FORMAT] [APP [APP ...]]
Generate Open API Specification
positional arguments:
APP 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 If your
asyncio.web.Application is returned by a function, you can use the syntax:
my_package.my_module:my_app()
optional arguments:
-h, --help show this help message and exit
-b FILE, --base-oas-file FILE
A file that will be used as base to generate OAS
-o FILE, --output FILE
File to write the output
-f FORMAT, --format FORMAT
The output format, can be 'json' or 'yaml' (default is json)
.. _demo: https://github.com/Maillol/aiohttp-pydantic/tree/main/demo
.. _aiohttp view: https://docs.aiohttp.org/en/stable/web_quickstart.html#class-based-views
Raw data
{
"_id": null,
"home_page": "https://github.com/Maillol/aiohttp-pydantic",
"name": "aiohttp-pydantic",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "aiohttp, pydantic, annotations, validation",
"author": "Vincent Maillol",
"author_email": "vincent.maillol@gmail.com",
"download_url": null,
"platform": null,
"description": "Aiohttp pydantic - Aiohttp View to validate and parse request\n=============================================================\n\n.. image:: https://github.com/Maillol/aiohttp-pydantic/actions/workflows/install-package-and-test.yml/badge.svg\n :target: https://github.com/Maillol/aiohttp-pydantic/actions/workflows/install-package-and-test.yml\n :alt: CI Status\n\n.. image:: https://img.shields.io/pypi/v/aiohttp-pydantic\n :target: https://img.shields.io/pypi/v/aiohttp-pydantic\n :alt: Latest PyPI package version\n\n.. image:: https://codecov.io/gh/Maillol/aiohttp-pydantic/branch/main/graph/badge.svg\n :target: https://codecov.io/gh/Maillol/aiohttp-pydantic\n :alt: codecov.io status for master branch\n\nAiohttp pydantic is an `aiohttp view`_ to easily parse and validate request.\nYou define using the function annotations what your methods for handling HTTP verbs expects and Aiohttp pydantic parses the HTTP request\nfor you, validates the data, and injects that you want as parameters.\n\n\nFeatures:\n\n- Query string, request body, URL path and HTTP headers validation.\n- Open API Specification generation.\n\n\nHow to install\n--------------\n\n.. code-block:: bash\n\n $ pip install aiohttp_pydantic\n\nExample:\n--------\n\n.. code-block:: python3\n\n from typing import Optional\n\n from aiohttp import web\n from aiohttp_pydantic import PydanticView\n from pydantic import BaseModel\n\n # Use pydantic BaseModel to validate request body\n class ArticleModel(BaseModel):\n name: str\n nb_page: Optional[int]\n\n\n # Create your PydanticView and add annotations.\n class ArticleView(PydanticView):\n\n async def post(self, article: ArticleModel):\n return web.json_response({'name': article.name,\n 'number_of_page': article.nb_page})\n\n async def get(self, with_comments: bool=False):\n return web.json_response({'with_comments': with_comments})\n\n\n app = web.Application()\n app.router.add_view('/article', ArticleView)\n web.run_app(app)\n\n\n.. code-block:: bash\n\n $ curl -X GET http://127.0.0.1:8080/article?with_comments=a\n [\n {\n \"in\": \"query string\",\n \"loc\": [\n \"with_comments\"\n ],\n \"msg\": \"Input should be a valid boolean, unable to interpret input\",\n \"input\": \"a\",\n \"type\": \"bool_parsing\"\n }\n ]\n\n $ curl -X GET http://127.0.0.1:8080/article?with_comments=yes\n {\"with_comments\": true}\n\n $ curl -H \"Content-Type: application/json\" -X POST http://127.0.0.1:8080/article --data '{}'\n [\n {\n \"in\": \"body\",\n \"loc\": [\n \"name\"\n ],\n \"msg\": \"Field required\",\n \"input\": {},\n \"type\": \"missing\"\n },\n {\n \"in\": \"body\",\n \"loc\": [\n \"nb_page\"\n ],\n \"msg\": \"Field required\",\n \"input\": {},\n \"type\": \"missing\"\n }\n ]\n\n $ curl -H \"Content-Type: application/json\" -X POST http://127.0.0.1:8080/article --data '{\"name\": \"toto\", \"nb_page\": \"3\"}'\n {\"name\": \"toto\", \"number_of_page\": 3}\n\n\nExample using view function handler\n-----------------------------------\n\n.. code-block:: python3\n\n\n from typing import Optional\n\n from aiohttp import web\n from aiohttp_pydantic.decorator import inject_params\n from pydantic import BaseModel\n\n\n # Use pydantic BaseModel to validate request body\n class ArticleModel(BaseModel):\n name: str\n nb_page: Optional[int]\n\n\n # Create your function decorated by 'inject_params' and add annotations.\n @inject_params\n async def post(article: ArticleModel):\n return web.json_response({'name': article.name,\n 'number_of_page': article.nb_page})\n\n\n # If you need request\n @inject_params.and_request\n async def get(request, with_comments: bool = False):\n request.app[\"logger\"](\"OK\")\n return web.json_response({'with_comments': with_comments})\n\n\n app = web.Application()\n app[\"logger\"] = print\n app.router.add_post('/article', post)\n app.router.add_get('/article', get)\n web.run_app(app)\n\n\nAPI:\n----\n\nInject Path Parameters\n~~~~~~~~~~~~~~~~~~~~~~\n\nTo declare a path parameter, you must declare your argument as a `positional-only parameters`_:\n\n\nExample:\n\n.. code-block:: python3\n\n class AccountView(PydanticView):\n async def get(self, customer_id: str, account_id: str, /):\n ...\n\n app = web.Application()\n app.router.add_get('/customers/{customer_id}/accounts/{account_id}', AccountView)\n\nInject Query String Parameters\n~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n\nTo declare a query parameter, you must declare your argument as a simple argument:\n\n\n.. code-block:: python3\n\n class AccountView(PydanticView):\n async def get(self, customer_id: Optional[str] = None):\n ...\n\n app = web.Application()\n app.router.add_get('/customers', AccountView)\n\n\nA query string parameter is generally optional and we do not want to force the user to set it in the URL.\nIt's recommended to define a default value. It's possible to get a multiple value for the same parameter using\nthe List type\n\n.. code-block:: python3\n\n from typing import List\n from pydantic import Field\n\n class AccountView(PydanticView):\n async def get(self, tags: List[str] = Field(default_factory=list)):\n ...\n\n app = web.Application()\n app.router.add_get('/customers', AccountView)\n\n\nInject Request Body\n~~~~~~~~~~~~~~~~~~~\n\nTo declare a body parameter, you must declare your argument as a simple argument annotated with `pydantic Model`_.\n\n\n.. code-block:: python3\n\n class Customer(BaseModel):\n first_name: str\n last_name: str\n\n class CustomerView(PydanticView):\n async def post(self, customer: Customer):\n ...\n\n app = web.Application()\n app.router.add_view('/customers', CustomerView)\n\nInject HTTP headers\n~~~~~~~~~~~~~~~~~~~\n\nTo declare a HTTP headers parameter, you must declare your argument as a `keyword-only argument`_.\n\n\n.. code-block:: python3\n\n class CustomerView(PydanticView):\n async def get(self, *, authorization: str, expire_at: datetime):\n ...\n\n app = web.Application()\n app.router.add_view('/customers', CustomerView)\n\n\n.. _positional-only parameters: https://www.python.org/dev/peps/pep-0570/\n.. _pydantic Model: https://pydantic-docs.helpmanual.io/usage/models/\n.. _keyword-only argument: https://www.python.org/dev/peps/pep-3102/\n\n\nFile Upload\n-----------\n\nYou can receive files in addition to Pydantic data in your views. Here\u2019s an example of how to use it:\nUsage Example\n\nSuppose you want to create an API that accepts a book (with a title and a number of pages) as well as two files\nrepresenting the pages of the book. Here\u2019s how you can do it:\n\n.. code-block:: python3\n\n from aiohttp import web\n from aiohttp_pydantic import PydanticView\n from aiohttp_pydantic.uploaded_file import UploadedFile\n from pydantic import BaseModel\n\n class BookModel(BaseModel):\n title: str\n nb_page: int\n\n class BookAndUploadFileView(PydanticView):\n async def post(self, book: BookModel, page_1: UploadedFile, page_2: UploadedFile):\n content_1 = (await page_1.read()).decode(\"utf-8\")\n content_2 = (await page_2.read()).decode(\"utf-8\")\n return web.json_response(\n {\"book\": book.model_dump(), \"content_1\": content_1, \"content_2\": content_2},\n status=201,\n )\n\nImplementation Details\n~~~~~~~~~~~~~~~~~~~~~~\n\nFiles are represented by instances of UploadedFile, which wrap an `aiohttp.BodyPartReader`_.\nUploadedFile exposes the read() and read_chunk() methods, allowing you to read the content of uploaded files asynchronously. You can use read() to get the complete content or read_chunk() to read chunks of data at a time.\n\nConstraints to Consider\n~~~~~~~~~~~~~~~~~~~~~~~\n\n1 - Argument Order: If you use both Pydantic models and UploadedFile, you must always define BaseModel\ntype arguments before UploadedFile type arguments. This ensures proper processing of the data.\n\n2 - File Reading Order: UploadedFile instances must be read in the order they are declared in the method.\nSince files are not pre-loaded in memory or on disk, it is important to respect this order.\nIf the reading order is not respected, a MultipartReadingError is raised.\n\n\n.. _aiohttp.BodyPartReader: https://docs.aiohttp.org/en/stable/multipart_reference.html#aiohttp.BodyPartReader\n\n\n\nAdd route to generate Open Api Specification (OAS)\n--------------------------------------------------\n\naiohttp_pydantic provides a sub-application to serve a route to generate Open Api Specification\nreading annotation in your PydanticView. Use *aiohttp_pydantic.oas.setup()* to add the sub-application\n\n.. code-block:: python3\n\n from aiohttp import web\n from aiohttp_pydantic import oas\n\n\n app = web.Application()\n oas.setup(app)\n\nBy default, the route to display the Open Api Specification is /oas but you can change it using\n*url_prefix* parameter\n\n\n.. code-block:: python3\n\n oas.setup(app, url_prefix='/spec-api')\n\nIf you want generate the Open Api Specification from specific aiohttp sub-applications.\non the same route, you must use *apps_to_expose* parameter.\n\n\n.. code-block:: python3\n\n from aiohttp import web\n from aiohttp_pydantic import oas\n\n app = web.Application()\n sub_app_1 = web.Application()\n sub_app_2 = web.Application()\n\n oas.setup(app, apps_to_expose=[sub_app_1, sub_app_2])\n\n\nYou can change the title or the version of the generated open api specification using\n*title_spec* and *version_spec* parameters:\n\n\n.. code-block:: python3\n\n oas.setup(app, title_spec=\"My application\", version_spec=\"1.2.3\")\n\n\nAdd annotation to define response content\n~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n\nThe module aiohttp_pydantic.oas.typing provides class to annotate a\nresponse content.\n\nFor example *r200[List[Pet]]* means the server responses with\nthe status code 200 and the response content is a List of Pet where Pet will be\ndefined using a pydantic.BaseModel\n\nThe docstring of methods will be parsed to fill the descriptions in the\nOpen Api Specification.\n\n\n.. code-block:: python3\n\n from aiohttp_pydantic import PydanticView\n from aiohttp_pydantic.oas.typing import r200, r201, r204, r404\n\n\n class Pet(BaseModel):\n id: int\n name: str\n\n\n class Error(BaseModel):\n error: str\n\n\n class PetCollectionView(PydanticView):\n async def get(self) -> r200[List[Pet]]:\n \"\"\"\n Find all pets\n\n Tags: pet\n \"\"\"\n pets = self.request.app[\"model\"].list_pets()\n return web.json_response([pet.dict() for pet in pets])\n\n async def post(self, pet: Pet) -> r201[Pet]:\n \"\"\"\n Add a new pet to the store\n\n Tags: pet\n Status Codes:\n 201: The pet is created\n \"\"\"\n self.request.app[\"model\"].add_pet(pet)\n return web.json_response(pet.dict())\n\n\n class PetItemView(PydanticView):\n async def get(self, id: int, /) -> Union[r200[Pet], r404[Error]]:\n \"\"\"\n Find a pet by ID\n\n Tags: pet\n Status Codes:\n 200: Successful operation\n 404: Pet not found\n \"\"\"\n pet = self.request.app[\"model\"].find_pet(id)\n return web.json_response(pet.dict())\n\n async def put(self, id: int, /, pet: Pet) -> r200[Pet]:\n \"\"\"\n Update an existing pet\n\n Tags: pet\n Status Codes:\n 200: successful operation\n \"\"\"\n self.request.app[\"model\"].update_pet(id, pet)\n return web.json_response(pet.dict())\n\n async def delete(self, id: int, /) -> r204:\n self.request.app[\"model\"].remove_pet(id)\n return web.Response(status=204)\n\n\nGroup parameters\n----------------\n\nIf your method has lot of parameters you can group them together inside one or several Groups.\n\n\n.. code-block:: python3\n\n from aiohttp_pydantic.injectors import Group\n\n class Pagination(Group):\n page_num: int = 1\n page_size: int = 15\n\n\n class ArticleView(PydanticView):\n\n async def get(self, page: Pagination):\n articles = Article.get(page.page_num, page.page_size)\n ...\n\n\nThe parameters page_num and page_size are expected in the query string, and\nset inside a Pagination object passed as page parameter.\n\nThe code above is equivalent to:\n\n\n.. code-block:: python3\n\n class ArticleView(PydanticView):\n\n async def get(self, page_num: int = 1, page_size: int = 15):\n articles = Article.get(page_num, page_size)\n ...\n\n\nYou can add methods or properties to your Group.\n\n\n.. code-block:: python3\n\n class Pagination(Group):\n page_num: int = 1\n page_size: int = 15\n\n @property\n def num(self):\n return self.page_num\n\n @property\n def size(self):\n return self.page_size\n\n def slice(self):\n return slice(self.num, self.size)\n\n\n class ArticleView(PydanticView):\n\n async def get(self, page: Pagination):\n articles = Article.get(page.num, page.size)\n ...\n\n\nCustom Validation error\n-----------------------\n\nYou can redefine the on_validation_error hook in your PydanticView\n\n.. code-block:: python3\n\n class PetView(PydanticView):\n\n async def on_validation_error(self,\n exception: ValidationError,\n context: str):\n errors = exception.errors()\n for error in errors:\n error[\"in\"] = context # context is \"body\", \"headers\", \"path\" or \"query string\"\n error[\"custom\"] = \"your custom field ...\"\n return json_response(data=errors, status=400)\n\n\n\nIf you use function based view:\n\n.. code-block:: python3\n\n\n async def custom_error(exception: ValidationError,\n context: str):\n errors = exception.errors()\n for error in errors:\n error[\"in\"] = context # context is \"body\", \"headers\", \"path\" or \"query string\"\n error[\"custom\"] = \"your custom field ...\"\n return json_response(data=errors, status=400)\n\n\n @inject_params(on_validation_error=custom_error)\n async def get(with_comments: bool = False):\n ...\n\n @inject_params.and_request(on_validation_error=custom_error)\n async def get(request, with_comments: bool = False):\n ...\n\n\nA tip to use the same error handling on each view\n\n\n.. code-block:: python3\n\n inject_params = inject_params(on_validation_error=custom_error)\n\n\n @inject_params\n async def post(article: ArticleModel):\n return web.json_response({'name': article.name,\n 'number_of_page': article.nb_page})\n\n\n @inject_params.and_request\n async def get(request, with_comments: bool = False):\n return web.json_response({'with_comments': with_comments})\n\n\n\n\nAdd security to the endpoints\n-----------------------------\n\naiohttp_pydantic provides a basic way to add security to the endpoints. You can define the security\non the setup level using the *security* parameter and then mark view methods that will require this security schema.\n\n.. code-block:: python3\n\n from aiohttp import web\n from aiohttp_pydantic import oas\n\n\n app = web.Application()\n oas.setup(app, security={\"APIKeyHeader\": {\"type\": \"apiKey\", \"in\": \"header\", \"name\": \"Authorization\"}})\n\n\nAnd then mark the view method with the *security* descriptor\n\n\n.. code-block:: python3\n\n\n from aiohttp_pydantic import PydanticView\n from aiohttp_pydantic.oas.typing import r200, r201, r204, r404\n\n\n class Pet(BaseModel):\n id: int\n name: str\n\n\n class Error(BaseModel):\n error: str\n\n\n class PetCollectionView(PydanticView):\n async def get(self) -> r200[List[Pet]]:\n \"\"\"\n Find all pets\n\n Security: APIKeyHeader\n Tags: pet\n \"\"\"\n pets = self.request.app[\"model\"].list_pets()\n return web.json_response([pet.dict() for pet in pets])\n\n async def post(self, pet: Pet) -> r201[Pet]:\n \"\"\"\n Add a new pet to the store\n\n Tags: pet\n Status Codes:\n 201: The pet is created\n \"\"\"\n self.request.app[\"model\"].add_pet(pet)\n return web.json_response(pet.dict())\n\n\nDemo\n----\n\nHave a look at `demo`_ for a complete example\n\n.. code-block:: bash\n\n git clone https://github.com/Maillol/aiohttp-pydantic.git\n cd aiohttp-pydantic\n pip install .\n python -m demo\n\nGo to http://127.0.0.1:8080/oas\n\nYou can generate the OAS in a json or yaml file using the aiohttp_pydantic.oas command:\n\n.. code-block:: bash\n\n python -m aiohttp_pydantic.oas demo.main\n\n.. code-block:: bash\n\n $ python3 -m aiohttp_pydantic.oas --help\n usage: __main__.py [-h] [-b FILE] [-o FILE] [-f FORMAT] [APP [APP ...]]\n\n Generate Open API Specification\n\n positional arguments:\n APP The name of the module containing the asyncio.web.Application. By default the variable named\n 'app' is loaded but you can define an other variable name ending the name of module with :\n characters and the name of variable. Example: my_package.my_module:my_app If your\n asyncio.web.Application is returned by a function, you can use the syntax:\n my_package.my_module:my_app()\n\n optional arguments:\n -h, --help show this help message and exit\n -b FILE, --base-oas-file FILE\n A file that will be used as base to generate OAS\n -o FILE, --output FILE\n File to write the output\n -f FORMAT, --format FORMAT\n The output format, can be 'json' or 'yaml' (default is json)\n\n\n.. _demo: https://github.com/Maillol/aiohttp-pydantic/tree/main/demo\n.. _aiohttp view: https://docs.aiohttp.org/en/stable/web_quickstart.html#class-based-views\n",
"bugtrack_url": null,
"license": null,
"summary": "Aiohttp View using pydantic to validate request body and query sting regarding method annotations.",
"version": "2.4.0",
"project_urls": {
"Homepage": "https://github.com/Maillol/aiohttp-pydantic"
},
"split_keywords": [
"aiohttp",
" pydantic",
" annotations",
" validation"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "5fd5845dab04d64039f4bb4fc657a2f81908fdc5f2fdb6cc9dacdf943d00e295",
"md5": "065ad37b8fd93239c305134677795ab1",
"sha256": "717c4d903a6c9b7715585b195abbb46715600c342c14ca747b7c4e2e2a23de68"
},
"downloads": -1,
"filename": "aiohttp_pydantic-2.4.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "065ad37b8fd93239c305134677795ab1",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 26845,
"upload_time": "2024-11-02T18:37:02",
"upload_time_iso_8601": "2024-11-02T18:37:02.363476Z",
"url": "https://files.pythonhosted.org/packages/5f/d5/845dab04d64039f4bb4fc657a2f81908fdc5f2fdb6cc9dacdf943d00e295/aiohttp_pydantic-2.4.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-11-02 18:37:02",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Maillol",
"github_project": "aiohttp-pydantic",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "aiohttp-pydantic"
}
JSON
