Compare commits

1 Commits

Author SHA1 Message Date
Vincent Maillol
2a064a75d9 update version 2020-10-25 20:52:04 +01:00
23 changed files with 93 additions and 585 deletions

View File

@@ -145,8 +145,8 @@ To declare a HTTP headers parameters, you must declare your argument as a `keywo
.. _pydantic Model: https://pydantic-docs.helpmanual.io/usage/models/
.. _keyword-only argument: https://www.python.org/dev/peps/pep-3102/
Add route to generate Open Api Specification (OAS)
--------------------------------------------------
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
@@ -168,8 +168,8 @@ By default, the route to display the Open Api Specification is /oas but you can
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.
If you want generate the Open Api Specification from several aiohttp sub-application.
on the same route, you must use *apps_to_expose* parameters
.. code-block:: python3
@@ -179,61 +179,11 @@ on the same route, you must use *apps_to_expose* parameter.
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
.. 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]]:
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)
oas.setup(app, apps_to_expose=[app, sub_app_1])
Demo
----
====
Have a look at `demo`_ for a complete example
@@ -246,11 +196,6 @@ Have a look at `demo`_ for a complete example
Go to http://127.0.0.1:8080/oas
You can generate the OAS in a json file using the command:
.. code-block:: bash
python -m aiohttp_pydantic.oas demo.main
.. _demo: https://github.com/Maillol/aiohttp-pydantic/tree/main/demo

View File

@@ -1,5 +1,5 @@
from .view import PydanticView
__version__ = "1.4.0"
__version__ = "1.1.0"
__all__ = ("PydanticView", "__version__")

View File

@@ -1,9 +1,11 @@
import abc
from inspect import signature
from typing import Callable, Tuple
from aiohttp.web_request import BaseRequest
from pydantic import BaseModel
from inspect import signature
import abc
class AbstractInjector(metaclass=abc.ABCMeta):

View File

@@ -1,11 +1,10 @@
from importlib import resources
from typing import Iterable
from importlib import resources
import jinja2
from aiohttp import web
from swagger_ui_bundle import swagger_ui_path
from .view import get_oas, oas_ui
from swagger_ui_bundle import swagger_ui_path
def setup(

View File

@@ -1,8 +0,0 @@
import argparse
from .cmd import setup
parser = argparse.ArgumentParser(description="Generate Open API Specification")
setup(parser)
args = parser.parse_args()
args.func(args)

View File

@@ -1,45 +0,0 @@
import argparse
import importlib
import json
from .view import generate_oas
def application_type(value):
"""
Return aiohttp application defined in the value.
"""
try:
module_name, app_name = value.split(":")
except ValueError:
module_name, app_name = value, "app"
module = importlib.import_module(module_name)
try:
if app_name.endswith("()"):
app_name = app_name.strip("()")
factory_app = getattr(module, app_name)
return factory_app()
return getattr(module, app_name)
except AttributeError as error:
raise argparse.ArgumentTypeError(error) from error
def setup(parser: argparse.ArgumentParser):
parser.add_argument(
"apps",
metavar="APP",
type=application_type,
nargs="*",
help="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",
)
parser.set_defaults(func=show_oas)
def show_oas(args: argparse.Namespace):
print(json.dumps(generate_oas(args.apps), sort_keys=True, indent=4))

View File

@@ -1,6 +1,3 @@
from typing import Union
class Info:
def __init__(self, spec: dict):
self._spec = spec.setdefault("info", {})
@@ -118,39 +115,6 @@ class Parameters:
return Parameter(spec)
class Response:
def __init__(self, spec: dict):
self._spec = spec
@property
def description(self) -> str:
return self._spec["description"]
@description.setter
def description(self, description: str):
self._spec["description"] = description
@property
def content(self):
return self._spec["content"]
@content.setter
def content(self, content: dict):
self._spec["content"] = content
class Responses:
def __init__(self, spec: dict):
self._spec = spec.setdefault("responses", {})
def __getitem__(self, status_code: Union[int, str]) -> Response:
if not (100 <= int(status_code) < 600):
raise ValueError("status_code must be between 100 and 599")
spec = self._spec.setdefault(str(status_code), {})
return Response(spec)
class OperationObject:
def __init__(self, spec: dict):
self._spec = spec
@@ -179,10 +143,6 @@ class OperationObject:
def parameters(self) -> Parameters:
return Parameters(self._spec)
@property
def responses(self) -> Responses:
return Responses(self._spec)
class PathItem:
def __init__(self, spec: dict):

View File

@@ -1,47 +0,0 @@
"""
This module provides type to annotate the content of web.Response returned by
the HTTP handlers.
The type are: r100, r101, ..., r599
Example:
class PetCollectionView(PydanticView):
async def get(self) -> Union[r200[List[Pet]], r404]:
...
"""
from functools import lru_cache
from types import new_class
from typing import Protocol, TypeVar
RespContents = TypeVar("RespContents", covariant=True)
_status_code = frozenset(f"r{code}" for code in range(100, 600))
@lru_cache(maxsize=len(_status_code))
def _make_status_code_type(status_code):
if status_code in _status_code:
return new_class(status_code, (Protocol[RespContents],))
def is_status_code_type(obj):
"""
Return True if obj is a status code type such as _200 or _404.
"""
name = getattr(obj, "__name__", None)
if name not in _status_code:
return False
return obj is _make_status_code_type(name)
def __getattr__(name):
if (status_code_type := _make_status_code_type(name)) is None:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
return status_code_type
__all__ = list(_status_code)
__all__.append("is_status_code_type")

View File

@@ -1,116 +1,51 @@
import typing
from typing import List, Type
from aiohttp.web import Response, json_response
from aiohttp.web_app import Application
from pydantic import BaseModel
from aiohttp.web import json_response, Response
from aiohttp_pydantic.oas.struct import OpenApiSpec3, OperationObject, PathItem
from typing import Type
from ..injectors import _parse_func_signature
from ..view import PydanticView, is_pydantic_view
from .typing import is_status_code_type
JSON_SCHEMA_TYPES = {float: "number", str: "string", int: "integer"}
def _is_pydantic_base_model(obj):
"""
Return true is obj is a pydantic.BaseModel subclass.
"""
try:
return issubclass(obj, BaseModel)
except TypeError:
return False
class _OASResponseBuilder:
"""
Parse the type annotated as returned by a function and
generate the OAS operation response.
"""
def __init__(self, oas_operation):
self._oas_operation = oas_operation
@staticmethod
def _handle_pydantic_base_model(obj):
if _is_pydantic_base_model(obj):
return obj.schema()
return {}
def _handle_list(self, obj):
if typing.get_origin(obj) is list:
return {
"type": "array",
"items": self._handle_pydantic_base_model(typing.get_args(obj)[0]),
}
return self._handle_pydantic_base_model(obj)
def _handle_status_code_type(self, obj):
if is_status_code_type(typing.get_origin(obj)):
status_code = typing.get_origin(obj).__name__[1:]
self._oas_operation.responses[status_code].content = {
"application/json": {
"schema": self._handle_list(typing.get_args(obj)[0])
}
}
elif is_status_code_type(obj):
status_code = obj.__name__[1:]
self._oas_operation.responses[status_code].content = {}
def _handle_union(self, obj):
if typing.get_origin(obj) is typing.Union:
for arg in typing.get_args(obj):
self._handle_status_code_type(arg)
self._handle_status_code_type(obj)
def build(self, obj):
self._handle_union(obj)
def _add_http_method_to_oas(
oas_path: PathItem, http_method: str, view: Type[PydanticView]
):
http_method = http_method.lower()
oas_operation: OperationObject = getattr(oas_path, http_method)
handler = getattr(view, http_method)
def _add_http_method_to_oas(oas_path: PathItem, method: str, view: Type[PydanticView]):
method = method.lower()
mtd: OperationObject = getattr(oas_path, method)
handler = getattr(view, method)
path_args, body_args, qs_args, header_args = _parse_func_signature(handler)
if body_args:
oas_operation.request_body.content = {
mtd.request_body.content = {
"application/json": {"schema": next(iter(body_args.values())).schema()}
}
i = 0
for i, (name, type_) in enumerate(path_args.items()):
oas_operation.parameters[i].required = True
oas_operation.parameters[i].in_ = "path"
oas_operation.parameters[i].name = name
oas_operation.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
mtd.parameters[i].required = True
mtd.parameters[i].in_ = "path"
mtd.parameters[i].name = name
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
for i, (name, type_) in enumerate(qs_args.items(), i + 1):
oas_operation.parameters[i].required = False
oas_operation.parameters[i].in_ = "query"
oas_operation.parameters[i].name = name
oas_operation.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
mtd.parameters[i].required = False
mtd.parameters[i].in_ = "query"
mtd.parameters[i].name = name
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
for i, (name, type_) in enumerate(header_args.items(), i + 1):
oas_operation.parameters[i].required = False
oas_operation.parameters[i].in_ = "header"
oas_operation.parameters[i].name = name
oas_operation.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
return_type = handler.__annotations__.get("return")
if return_type is not None:
_OASResponseBuilder(oas_operation).build(return_type)
mtd.parameters[i].required = False
mtd.parameters[i].in_ = "header"
mtd.parameters[i].name = name
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
def generate_oas(apps: List[Application]) -> dict:
async def get_oas(request):
"""
Generate and return Open Api Specification from PydanticView in application.
Generate Open Api Specification from PydanticView in application.
"""
apps = request.app["apps to expose"]
oas = OpenApiSpec3()
for app in apps:
for resources in app.router.resources():
@@ -125,15 +60,7 @@ def generate_oas(apps: List[Application]) -> dict:
else:
_add_http_method_to_oas(path, resource_route.method, view)
return oas.spec
async def get_oas(request):
"""
View to generate the Open Api Specification from PydanticView in application.
"""
apps = request.app["apps to expose"]
return json_response(generate_oas(apps))
return json_response(oas.spec)
async def oas_ui(request):

View File

@@ -1,20 +1,20 @@
from functools import update_wrapper
from inspect import iscoroutinefunction
from typing import Any, Callable, Generator, Iterable
from aiohttp.abc import AbstractView
from aiohttp.hdrs import METH_ALL
from aiohttp.web import json_response
from aiohttp.web_exceptions import HTTPMethodNotAllowed
from aiohttp.web_response import StreamResponse
from pydantic import ValidationError
from typing import Generator, Any, Callable, Type, Iterable
from aiohttp.web import json_response
from functools import update_wrapper
from .injectors import (
AbstractInjector,
BodyGetter,
HeadersGetter,
MatchInfoGetter,
HeadersGetter,
QueryGetter,
BodyGetter,
AbstractInjector,
_parse_func_signature,
)

View File

@@ -1,5 +1,25 @@
from aiohttp import web
from .main import app
from aiohttp_pydantic import oas
from aiohttp.web import middleware
from .view import PetItemView, PetCollectionView
from .model import Model
@middleware
async def pet_not_found_to_404(request, handler):
try:
return await handler(request)
except Model.NotFound as key:
return web.json_response({"error": f"Pet {key} does not exist"}, status=404)
app = web.Application(middlewares=[pet_not_found_to_404])
oas.setup(app)
app["model"] = Model()
app.router.add_view("/pets", PetCollectionView)
app.router.add_view("/pets/{id}", PetItemView)
web.run_app(app)

View File

@@ -1,22 +0,0 @@
from aiohttp.web import Application, json_response, middleware
from aiohttp_pydantic import oas
from .model import Model
from .view import PetCollectionView, PetItemView
@middleware
async def pet_not_found_to_404(request, handler):
try:
return await handler(request)
except Model.NotFound as key:
return json_response({"error": f"Pet {key} does not exist"}, status=404)
app = Application(middlewares=[pet_not_found_to_404])
oas.setup(app)
app["model"] = Model()
app.router.add_view("/pets", PetCollectionView)
app.router.add_view("/pets/{id}", PetItemView)

View File

@@ -6,10 +6,6 @@ class Pet(BaseModel):
name: str
class Error(BaseModel):
error: str
class Model:
"""
To keep simple this demo, we use a simple dict as database to

View File

@@ -1,32 +1,28 @@
from typing import List, Union
from aiohttp_pydantic import PydanticView
from aiohttp import web
from aiohttp_pydantic import PydanticView
from aiohttp_pydantic.oas.typing import r200, r201, r204, r404
from .model import Error, Pet
from .model import Pet
class PetCollectionView(PydanticView):
async def get(self) -> r200[List[Pet]]:
async def get(self):
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]:
async def post(self, pet: 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]]:
async def get(self, id: int, /):
pet = self.request.app["model"].find_pet(id)
return web.json_response(pet.dict())
async def put(self, id: int, /, pet: Pet) -> r200[Pet]:
async def put(self, id: int, /, pet: Pet):
self.request.app["model"].update_pet(id, pet)
return web.json_response(pet.dict())
async def delete(self, id: int, /) -> r204:
async def delete(self, id: int, /):
self.request.app["model"].remove_pet(id)
return web.Response(status=204)
return web.json_response(id)

View File

@@ -1,27 +0,0 @@
from aiohttp import web
from aiohttp_pydantic import PydanticView
class View1(PydanticView):
async def get(self, a: int, /):
return web.json_response()
class View2(PydanticView):
async def post(self, b: int, /):
return web.json_response()
sub_app = web.Application()
sub_app.router.add_view("/route-2/{b}", View2)
app = web.Application()
app.router.add_view("/route-1/{a}", View1)
app.add_subapp("/sub-app", sub_app)
def make_app():
app = web.Application()
app.router.add_view("/route-3/{a}", View1)
return app

View File

@@ -1,120 +0,0 @@
import argparse
from textwrap import dedent
import pytest
from aiohttp_pydantic.oas import cmd
@pytest.fixture
def cmd_line():
parser = argparse.ArgumentParser()
cmd.setup(parser)
return parser
def test_show_oad_of_app(cmd_line, capfd):
args = cmd_line.parse_args(["tests.test_oas.test_cmd.sample"])
args.func(args)
captured = capfd.readouterr()
expected = dedent(
"""
{
"openapi": "3.0.0",
"paths": {
"/route-1/{a}": {
"get": {
"parameters": [
{
"in": "path",
"name": "a",
"required": true,
"schema": {
"type": "integer"
}
}
]
}
},
"/sub-app/route-2/{b}": {
"post": {
"parameters": [
{
"in": "path",
"name": "b",
"required": true,
"schema": {
"type": "integer"
}
}
]
}
}
}
}
"""
)
assert captured.out.strip() == expected.strip()
def test_show_oad_of_sub_app(cmd_line, capfd):
args = cmd_line.parse_args(["tests.test_oas.test_cmd.sample:sub_app"])
args.func(args)
captured = capfd.readouterr()
expected = dedent(
"""
{
"openapi": "3.0.0",
"paths": {
"/sub-app/route-2/{b}": {
"post": {
"parameters": [
{
"in": "path",
"name": "b",
"required": true,
"schema": {
"type": "integer"
}
}
]
}
}
}
}
"""
)
assert captured.out.strip() == expected.strip()
def test_show_oad_of_a_callable(cmd_line, capfd):
args = cmd_line.parse_args(["tests.test_oas.test_cmd.sample:make_app()"])
args.func(args)
captured = capfd.readouterr()
expected = dedent(
"""
{
"openapi": "3.0.0",
"paths": {
"/route-3/{a}": {
"get": {
"parameters": [
{
"in": "path",
"name": "a",
"required": true,
"schema": {
"type": "integer"
}
}
]
}
}
}
}
"""
)
assert captured.out.strip() == expected.strip()

View File

@@ -1,11 +1,8 @@
from typing import List, Union
from pydantic.main import BaseModel
from aiohttp_pydantic import PydanticView, oas
from aiohttp import web
import pytest
from aiohttp import web
from pydantic.main import BaseModel
from aiohttp_pydantic import PydanticView, oas
from aiohttp_pydantic.oas.typing import r200, r201, r204, r404
class Pet(BaseModel):
@@ -14,21 +11,21 @@ class Pet(BaseModel):
class PetCollectionView(PydanticView):
async def get(self) -> r200[List[Pet]]:
async def get(self):
return web.json_response()
async def post(self, pet: Pet) -> r201[Pet]:
async def post(self, pet: Pet):
return web.json_response()
class PetItemView(PydanticView):
async def get(self, id: int, /) -> Union[r200[Pet], r404]:
async def get(self, id: int, /):
return web.json_response()
async def put(self, id: int, /, pet: Pet):
return web.json_response()
async def delete(self, id: int, /) -> r204:
async def delete(self, id: int, /):
return web.json_response()
@@ -51,28 +48,7 @@ async def test_generated_oas_should_have_pets_paths(generated_oas):
async def test_pets_route_should_have_get_method(generated_oas):
assert generated_oas["paths"]["/pets"]["get"] == {
"responses": {
"200": {
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"title": "Pet",
"type": "object",
"properties": {
"id": {"title": "Id", "type": "integer"},
"name": {"title": "Name", "type": "string"},
},
"required": ["id", "name"],
},
}
}
}
}
}
}
assert generated_oas["paths"]["/pets"]["get"] == {}
async def test_pets_route_should_have_post_method(generated_oas):
@@ -81,34 +57,17 @@ async def test_pets_route_should_have_post_method(generated_oas):
"content": {
"application/json": {
"schema": {
"title": "Pet",
"type": "object",
"properties": {
"id": {"title": "Id", "type": "integer"},
"name": {"title": "Name", "type": "string"},
},
"required": ["id", "name"],
"title": "Pet",
"type": "object",
}
}
}
},
"responses": {
"201": {
"content": {
"application/json": {
"schema": {
"title": "Pet",
"type": "object",
"properties": {
"id": {"title": "Id", "type": "integer"},
"name": {"title": "Name", "type": "string"},
},
"required": ["id", "name"],
}
}
}
}
},
}
}
@@ -120,13 +79,12 @@ async def test_pets_id_route_should_have_delete_method(generated_oas):
assert generated_oas["paths"]["/pets/{id}"]["delete"] == {
"parameters": [
{
"required": True,
"in": "path",
"name": "id",
"required": True,
"schema": {"type": "integer"},
}
],
"responses": {"204": {"content": {}}},
]
}
@@ -139,25 +97,7 @@ async def test_pets_id_route_should_have_get_method(generated_oas):
"required": True,
"schema": {"type": "integer"},
}
],
"responses": {
"200": {
"content": {
"application/json": {
"schema": {
"properties": {
"id": {"title": "Id", "type": "integer"},
"name": {"title": "Name", "type": "string"},
},
"required": ["id", "name"],
"title": "Pet",
"type": "object",
}
}
}
},
"404": {"content": {}},
},
]
}

View File

@@ -1,8 +1,6 @@
from uuid import UUID
from pydantic import BaseModel
from aiohttp_pydantic.injectors import _parse_func_signature
from pydantic import BaseModel
from uuid import UUID
class User(BaseModel):

View File

@@ -1,8 +1,6 @@
from typing import Optional
from aiohttp import web
from pydantic import BaseModel
from typing import Optional
from aiohttp import web
from aiohttp_pydantic import PydanticView

View File

@@ -1,9 +1,7 @@
import json
from datetime import datetime
from aiohttp import web
from aiohttp_pydantic import PydanticView
from datetime import datetime
import json
class JSONEncoder(json.JSONEncoder):

View File

@@ -1,5 +1,4 @@
from aiohttp import web
from aiohttp_pydantic import PydanticView

View File

@@ -1,5 +1,4 @@
from aiohttp import web
from aiohttp_pydantic import PydanticView