Compare commits

1 Commits

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

3
.gitignore vendored
View File

@@ -1,9 +1,6 @@
.coverage
.idea/
.pytest_cache
__pycache__
aiohttp_pydantic.egg-info/
build/
coverage.xml
dist/

View File

@@ -2,14 +2,11 @@ language: python
python:
- '3.8'
script:
- pytest --cov-report=xml --cov=aiohttp_pydantic tests/
- pytest tests/
install:
- pip install -U setuptools wheel pip
- pip install -r requirements/test.txt
- pip install -r requirements/ci.txt
- pip install -r test_requirements.txt
- pip install .
after_success:
- codecov
deploy:
provider: pypi
username: __token__
@@ -19,4 +16,4 @@ deploy:
on:
tags: true
branch: main
python: '3.8'
python: '3.8'

View File

@@ -1,28 +1,6 @@
Aiohttp pydantic - Aiohttp View to validate and parse request
=============================================================
.. image:: https://travis-ci.org/Maillol/aiohttp-pydantic.svg?branch=main
:target: https://travis-ci.org/Maillol/aiohttp-pydantic
.. 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
--------------
@@ -167,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
@@ -190,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
@@ -201,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
@@ -268,12 +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
.. _aiohttp view: https://docs.aiohttp.org/en/stable/web_quickstart.html#class-based-views

View File

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

View File

@@ -1,13 +1,11 @@
import abc
from inspect import signature
from json.decoder import JSONDecodeError
from typing import Callable, Tuple
from aiohttp.web_exceptions import HTTPBadRequest
from aiohttp.web_request import BaseRequest
from pydantic import BaseModel
from inspect import signature
from .utils import is_pydantic_base_model
import abc
class AbstractInjector(metaclass=abc.ABCMeta):
@@ -49,13 +47,7 @@ class BodyGetter(AbstractInjector):
self.arg_name, self.model = next(iter(args_spec.items()))
async def inject(self, request: BaseRequest, args_view: list, kwargs_view: dict):
try:
body = await request.json()
except JSONDecodeError:
raise HTTPBadRequest(
text='{"error": "Malformed JSON"}', content_type="application/json"
)
body = await request.json()
kwargs_view[self.arg_name] = self.model(**body)
@@ -108,7 +100,7 @@ def _parse_func_signature(func: Callable) -> Tuple[dict, dict, dict, dict]:
if param_spec.kind is param_spec.POSITIONAL_ONLY:
path_args[param_name] = param_spec.annotation
elif param_spec.kind is param_spec.POSITIONAL_OR_KEYWORD:
if is_pydantic_base_model(param_spec.annotation):
if issubclass(param_spec.annotation, BaseModel):
body_args[param_name] = param_spec.annotation
else:
qs_args[param_name] = param_spec.annotation

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,17 +1,10 @@
"""
Utility to write Open Api Specifications using the Python language.
"""
from typing import Union
class Info:
def __init__(self, spec: dict):
self._spec = spec.setdefault("info", {})
@property
def title(self):
return self._spec.get("title")
return self._spec["title"]
@title.setter
def title(self, title):
@@ -19,7 +12,7 @@ class Info:
@property
def description(self):
return self._spec.get("description")
return self._spec["description"]
@description.setter
def description(self, description):
@@ -27,20 +20,12 @@ class Info:
@property
def version(self):
return self._spec.get("version")
return self._spec["version"]
@version.setter
def version(self, version):
self._spec["version"] = version
@property
def terms_of_service(self):
return self._spec.get("termsOfService")
@terms_of_service.setter
def terms_of_service(self, terms_of_service):
self._spec["termsOfService"] = terms_of_service
class RequestBody:
def __init__(self, spec: dict):
@@ -55,8 +40,8 @@ class RequestBody:
self._spec["description"] = description
@property
def required(self) -> bool:
return self._spec.get("required", False)
def required(self):
return self._spec["required"]
@required.setter
def required(self, required: bool):
@@ -130,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
@@ -191,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):
@@ -232,22 +180,6 @@ class PathItem:
def trace(self) -> OperationObject:
return OperationObject(self._spec.setdefault("trace", {}))
@property
def description(self) -> str:
return self._spec["description"]
@description.setter
def description(self, description: str):
self._spec["description"] = description
@property
def summary(self) -> str:
return self._spec["summary"]
@summary.setter
def summary(self, summary: str):
self._spec["summary"] = summary
class Paths:
def __init__(self, spec: dict):
@@ -272,7 +204,7 @@ class Server:
@property
def description(self) -> str:
return self._spec["description"]
return self._spec["url"]
@description.setter
def description(self, description: str):
@@ -312,8 +244,3 @@ class OpenApiSpec3:
@property
def spec(self):
return self._spec
@property
def definitions(self):
self._spec.setdefault('definitions', {})
return self._spec['definitions']

View File

@@ -1,48 +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, Optional, Type
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],))
return None
def is_status_code_type(obj) -> bool:
"""
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,145 +1,52 @@
import typing
from datetime import date, datetime
from inspect import getdoc
from itertools import count
from typing import List, Type
from uuid import UUID
from aiohttp.web import Response, json_response
from aiohttp.web_app import Application
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 ..utils import is_pydantic_base_model
from ..view import PydanticView, is_pydantic_view
from .typing import is_status_code_type
JSON_SCHEMA_TYPES = {
float: {"type": "number"},
str: {"type": "string"},
int: {"type": "integer"},
UUID: {"type": "string", "format": "uuid"},
bool: {"type": "boolean"},
datetime: {"type": "string", "format": "date-time"},
date: {"type": "string", "format": "date"},
}
def _handle_optional(type_):
"""
Returns the type wrapped in Optional or None.
>>> _handle_optional(int)
>>> _handle_optional(Optional[str])
<class 'str'>
"""
if typing.get_origin(type_) is typing.Union:
args = typing.get_args(type_)
if len(args) == 2 and type(None) in args:
return next(iter(set(args) - {type(None)}))
return None
JSON_SCHEMA_TYPES = {float: "number", str: "string", int: "integer"}
class _OASResponseBuilder:
"""
Parse the type annotated as returned by a function and
generate the OAS operation response.
"""
def __init__(self, oas_operation, definitions):
self._oas_operation = oas_operation
self._definitions = definitions
def _process_definitions(self, schema):
if 'definitions' in schema:
for k, v in schema['definitions'].items():
self._definitions[k] = v
return {i:schema[i] for i in schema if i!='definitions'}
def _handle_pydantic_base_model(self, obj):
if is_pydantic_base_model(obj):
return self._process_definitions(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], definitions: dict
):
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)
description = getdoc(handler)
if description:
oas_operation.description = description
if body_args:
oas_operation.request_body.content = {
mtd.request_body.content = {
"application/json": {"schema": next(iter(body_args.values())).schema()}
}
indexes = count()
for args_location, args in (
("path", path_args.items()),
("query", qs_args.items()),
("header", header_args.items()),
):
for name, type_ in args:
i = next(indexes)
oas_operation.parameters[i].in_ = args_location
oas_operation.parameters[i].name = name
optional_type = _handle_optional(type_)
if optional_type is None:
oas_operation.parameters[i].schema = JSON_SCHEMA_TYPES[type_]
oas_operation.parameters[i].required = True
else:
oas_operation.parameters[i].schema = JSON_SCHEMA_TYPES[optional_type]
oas_operation.parameters[i].required = False
i = 0
for i, (name, type_) in enumerate(path_args.items()):
mtd.parameters[i].required = True
mtd.parameters[i].in_ = "path"
mtd.parameters[i].name = name
mtd.parameters[i].schema = {"type": JSON_SCHEMA_TYPES[type_]}
return_type = handler.__annotations__.get("return")
if return_type is not None:
_OASResponseBuilder(oas_operation, definitions).build(return_type)
for i, (name, type_) in enumerate(qs_args.items(), i + 1):
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):
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():
for resource_route in resources:
@@ -149,19 +56,11 @@ def generate_oas(apps: List[Application]) -> dict:
path = oas.paths[info.get("path", info.get("formatter"))]
if resource_route.method == "*":
for method_name in view.allowed_methods:
_add_http_method_to_oas(path, method_name, view, oas.definitions)
_add_http_method_to_oas(path, method_name, view)
else:
_add_http_method_to_oas(path, resource_route.method, view, oas.definitions)
_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):
@@ -172,9 +71,6 @@ async def oas_ui(request):
static_url = request.app.router["static"].url_for(filename="")
spec_url = request.app.router["spec"].url_for()
if request.scheme != request.headers.get('x-forwarded-proto', request.scheme):
request = request.clone(scheme=request.headers['x-forwarded-proto'])
host = request.url.origin()
return Response(

View File

@@ -1,11 +0,0 @@
from pydantic import BaseModel
def is_pydantic_base_model(obj):
"""
Return true is obj is a pydantic.BaseModel subclass.
"""
try:
return issubclass(obj, BaseModel)
except TypeError:
return False

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

@@ -4,11 +4,6 @@ from pydantic import BaseModel
class Pet(BaseModel):
id: int
name: str
age: int
class Error(BaseModel):
error: str
class Model:

View File

@@ -1,34 +1,28 @@
from typing import List, Optional, 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, age: Optional[int] = None) -> r200[List[Pet]]:
async def get(self):
pets = self.request.app["model"].list_pets()
return web.json_response(
[pet.dict() for pet in pets if age is None or age == pet.age]
)
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,7 +0,0 @@
certifi==2020.11.8
chardet==3.0.4
codecov==2.1.10
coverage==5.3
idna==2.10
requests==2.25.0
urllib3==1.26.2

View File

@@ -1,13 +0,0 @@
attrs==20.3.0
coverage==5.3
iniconfig==1.1.1
packaging==20.4
pluggy==0.13.1
py==1.9.0
pyparsing==2.4.7
pytest==6.1.2
pytest-aiohttp==0.3.0
pytest-cov==2.10.1
six==1.15.0
toml==0.10.2
typing-extensions==3.7.4.3

View File

@@ -35,8 +35,8 @@ install_requires =
swagger-ui-bundle
[options.extras_require]
test = pytest==6.1.2; pytest-aiohttp==0.3.0; pytest-cov==2.10.1
ci = pytest==6.1.2; pytest-aiohttp==0.3.0; pytest-cov==2.10.1; codecov==2.1.10
test = pytest; pytest-aiohttp
[options.packages.find]
exclude =

3
test_requirements.txt Normal file
View File

@@ -0,0 +1,3 @@
pytest==6.1.1
pytest-aiohttp==0.3.0
typing_extensions>=3.6.5

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,54 +0,0 @@
import pytest
from aiohttp_pydantic.oas.struct import OpenApiSpec3
def test_info_title():
oas = OpenApiSpec3()
assert oas.info.title is None
oas.info.title = "Info Title"
assert oas.info.title == "Info Title"
assert oas.spec == {"info": {"title": "Info Title"}, "openapi": "3.0.0"}
def test_info_description():
oas = OpenApiSpec3()
assert oas.info.description is None
oas.info.description = "info description"
assert oas.info.description == "info description"
assert oas.spec == {"info": {"description": "info description"}, "openapi": "3.0.0"}
def test_info_version():
oas = OpenApiSpec3()
assert oas.info.version is None
oas.info.version = "3.14"
assert oas.info.version == "3.14"
assert oas.spec == {"info": {"version": "3.14"}, "openapi": "3.0.0"}
def test_info_terms_of_service():
oas = OpenApiSpec3()
assert oas.info.terms_of_service is None
oas.info.terms_of_service = "http://example.com/terms/"
assert oas.info.terms_of_service == "http://example.com/terms/"
assert oas.spec == {
"info": {"termsOfService": "http://example.com/terms/"},
"openapi": "3.0.0",
}
@pytest.mark.skip("Not yet implemented")
def test_info_license():
oas = OpenApiSpec3()
oas.info.license.name = "Apache 2.0"
oas.info.license.url = "https://www.apache.org/licenses/LICENSE-2.0.html"
assert oas.spec == {
"info": {
"license": {
"name": "Apache 2.0",
"url": "https://www.apache.org/licenses/LICENSE-2.0.html",
}
},
"openapi": "3.0.0",
}

View File

@@ -1,124 +0,0 @@
from aiohttp_pydantic.oas.struct import OpenApiSpec3
def test_paths_description():
oas = OpenApiSpec3()
oas.paths["/users/{id}"].description = "This route ..."
assert oas.spec == {
"openapi": "3.0.0",
"paths": {"/users/{id}": {"description": "This route ..."}},
}
def test_paths_get():
oas = OpenApiSpec3()
oas.paths["/users/{id}"].get
assert oas.spec == {"openapi": "3.0.0", "paths": {"/users/{id}": {"get": {}}}}
def test_paths_operation_description():
oas = OpenApiSpec3()
operation = oas.paths["/users/{id}"].get
operation.description = "Long descriptions ..."
assert oas.spec == {
"openapi": "3.0.0",
"paths": {"/users/{id}": {"get": {"description": "Long descriptions ..."}}},
}
def test_paths_operation_summary():
oas = OpenApiSpec3()
operation = oas.paths["/users/{id}"].get
operation.summary = "Updates a pet in the store with form data"
assert oas.spec == {
"openapi": "3.0.0",
"paths": {
"/users/{id}": {
"get": {"summary": "Updates a pet in the store with form data"}
}
},
}
def test_paths_operation_parameters():
oas = OpenApiSpec3()
operation = oas.paths["/users/{petId}"].get
parameter = operation.parameters[0]
parameter.name = "petId"
parameter.description = "ID of pet that needs to be updated"
parameter.in_ = "path"
parameter.required = True
assert oas.spec == {
"openapi": "3.0.0",
"paths": {
"/users/{petId}": {
"get": {
"parameters": [
{
"description": "ID of pet that needs to be updated",
"in": "path",
"name": "petId",
"required": True,
}
]
}
}
},
}
def test_paths_operation_requestBody():
oas = OpenApiSpec3()
request_body = oas.paths["/users/{petId}"].get.request_body
request_body.description = "user to add to the system"
request_body.content = {
"application/json": {
"schema": {"$ref": "#/components/schemas/User"},
"examples": {
"user": {
"summary": "User Example",
"externalValue": "http://foo.bar/examples/user-example.json",
}
},
}
}
request_body.required = True
assert oas.spec == {
"openapi": "3.0.0",
"paths": {
"/users/{petId}": {
"get": {
"requestBody": {
"content": {
"application/json": {
"examples": {
"user": {
"externalValue": "http://foo.bar/examples/user-example.json",
"summary": "User Example",
}
},
"schema": {"$ref": "#/components/schemas/User"},
}
},
"description": "user to add to the system",
"required": True,
}
}
}
},
}
def test_paths_operation_responses():
oas = OpenApiSpec3()
response = oas.paths["/users/{petId}"].get.responses[200]
response.description = "A complex object array response"
response.content = {
"application/json": {
"schema": {
"type": "array",
"items": {"$ref": "#/components/schemas/VeryComplexType"},
}
}
}

View File

@@ -1,36 +0,0 @@
import pytest
from aiohttp_pydantic.oas.struct import OpenApiSpec3
def test_sever_url():
oas = OpenApiSpec3()
oas.servers[0].url = "https://development.gigantic-server.com/v1"
oas.servers[1].url = "https://development.gigantic-server.com/v2"
assert oas.spec == {
"openapi": "3.0.0",
"servers": [
{"url": "https://development.gigantic-server.com/v1"},
{"url": "https://development.gigantic-server.com/v2"},
],
}
def test_sever_description():
oas = OpenApiSpec3()
oas.servers[0].url = "https://development.gigantic-server.com/v1"
oas.servers[0].description = "Development server"
assert oas.spec == {
"openapi": "3.0.0",
"servers": [
{
"url": "https://development.gigantic-server.com/v1",
"description": "Development server",
}
],
}
@pytest.mark.skip("Not yet implemented")
def test_sever_variables():
oas = OpenApiSpec3()

View File

@@ -1,12 +1,8 @@
from typing import List, Optional, Union
from uuid import UUID
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):
@@ -15,27 +11,21 @@ class Pet(BaseModel):
class PetCollectionView(PydanticView):
async def get(
self, format: str, name: Optional[str] = None, *, promo: Optional[UUID] = None
) -> r200[List[Pet]]:
"""
Get a list of pets
"""
async def get(self):
return web.json_response()
async def post(self, pet: Pet) -> r201[Pet]:
"""Create a 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()
@@ -58,86 +48,26 @@ 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"] == {
"description": "Get a list of pets",
"parameters": [
{
"in": "query",
"name": "format",
"required": True,
"schema": {"type": "string"},
},
{
"in": "query",
"name": "name",
"required": False,
"schema": {"type": "string"},
},
{
"in": "header",
"name": "promo",
"required": False,
"schema": {"format": "uuid", "type": "string"},
},
],
"responses": {
"200": {
"content": {
"application/json": {
"schema": {
"items": {
"properties": {
"id": {"title": "Id", "type": "integer"},
"name": {"title": "Name", "type": "string"},
},
"required": ["id", "name"],
"title": "Pet",
"type": "object",
},
"type": "array",
}
}
}
}
},
}
assert generated_oas["paths"]["/pets"]["get"] == {}
async def test_pets_route_should_have_post_method(generated_oas):
assert generated_oas["paths"]["/pets"]["post"] == {
"description": "Create a Pet",
"requestBody": {
"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"],
}
}
}
}
},
}
}
@@ -149,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": {}}},
]
}
@@ -168,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,13 +1,10 @@
from typing import Optional
from aiohttp import web
from aiohttp_pydantic import PydanticView
class ArticleView(PydanticView):
async def get(self, with_comments: bool, age: Optional[int] = None):
return web.json_response({"with_comments": with_comments, "age": age})
async def get(self, with_comments: bool):
return web.json_response({"with_comments": with_comments})
async def test_get_article_without_required_qs_should_return_an_error_message(
@@ -55,22 +52,7 @@ async def test_get_article_with_valid_qs_should_return_the_parsed_type(
app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
resp = await client.get("/article", params={"with_comments": "yes", "age": 3})
assert resp.status == 200
assert resp.content_type == "application/json"
assert await resp.json() == {"with_comments": True, "age": 3}
async def test_get_article_with_valid_qs_and_omitted_optional_should_return_none(
aiohttp_client, loop
):
app = web.Application()
app.router.add_view("/article", ArticleView)
client = await aiohttp_client(app)
resp = await client.get("/article", params={"with_comments": "yes"})
assert resp.status == 200
assert resp.content_type == "application/json"
assert await resp.json() == {"with_comments": True, "age": None}
assert await resp.json() == {"with_comments": True}