aiohttp-apischema¶
API schema generation and input validation for aiohttp.
Basic usage¶
First create a aiohttp_apischema.SchemaGenerator instance:
from aiohttp_apischema import SchemaGenerator
SCHEMA = SchemaGenerator()
Then decorate endpoint handlers with aiohttp_schema.SchemaGenerator.api()
which should be included in the schema and use aiohttp_schema.APIResponse for
the return type:
from aiohttp_apischema import APIResponse
@SCHEMA.api()
async def foo(request: web.Request) -> APIResponse[list[str], Literal[200]]:
return APIResponse(["foo"])
Or for Class Based Views
the aiohttp_apischema.SchemaGenerator.api_view() decorator can be used:
@SCHEMA.api_view()
class Handler(web.View):
async def get(self) -> APIResponse[int, Literal[200]] | APIResponse[None, Literal[404]]:
bar_id = int(self.request.match_info["id"])
if bar_id == 1:
return APIResponse(42)
return APIResponse[None, Literal[404]](None, status=404)
Then call the setup method when building your app:
app = web.Application()
app.router.add_get("/foo", foo)
app.router.add_view(r"/bar/{id:\d+}", Handler)
SCHEMA.setup(app)
You can now view the docs under the /swagger/ path.
Validation usage¶
Validation of the request body is achieved by adding a positional parameter:
async def handler(request: web.Request, body: dict[int, str]) -> APIResponse[int]:
# body has been validated, so we can be sure the keys are int now.
return APIResponse(sum(body.keys()))
This will include the information in the schema’s requestBody, plus it will validate the input from the user. If validation fails it will return a 400 response with information about what was incorrect.
Keyword-only parameters can be defined for query arguments:
class QueryArgs(TypedDict, total=False):
sort: Literal["asc", "desc"]
async def handler(request: web.Request, *, query: QueryArgs) -> APIResponse[int]:
return sorted(results, reverse=query.get("sort", "asc") == "desc")
Pydantic options¶
You can add custom Pydantic options using typing.Annotated:
class QueryArgs(TypedDict):
sort: Annotated[Literal["asc", "desc"], pydantic.Field(default="asc")]
async def handler(request: web.Request, *, query: QueryArgs) -> APIResponse[int]:
return sorted(results, reverse=query["sort"] == "desc")
Customising schema generation¶
Summary and description¶
You can use docstrings to customise the summary and description shown in the schema:
async def handler(request: web.Request) -> APIResponse[int, Literal[200]]:
"""This will appear as the summary.
This is a
longer
description.
"""
Library Installation¶
The aiohttp_apischema library can be installed with pip:
$ pip install aiohttp-apischema
Source code¶
The project is hosted on GitHub.
Please feel free to file an issue on bug tracker if you have found a bug or have some suggestion for library improvements.
License¶
aiohttp_apischema is offered under the Apache 2 license.
