maillol / aiohttp-pydantic Goto Github PK

View Code? Open in Web Editor NEW

60.0 60.0 19.0 141 KB

Aiohttp View that validates request body and query sting regarding the annotations declared in the View method

License: MIT License

Python 98.24% Jinja 1.76%

aiohttp-pydantic's People

Contributors

Stargazers

Watchers

Forkers

jar3b bengreen vakhet drderuiter reskov spinenkoia romkazor ffkirill danik-tro herrmuellerluedenscheid heroickatora steersbob khrebtukov zadewg codereverser tommmlij shizacat mbuczkowski-gm

aiohttp-pydantic's Issues

generate the OAS in json tips

Any chance I could get some tips on how to do this?

Copy paste from this repo:
python -m aiohttp_pydantic.oas demo.main

demo would be the name of the python package that was installed locally, .main does that refer to main.py or __main__.py?

For my app in order to generate the OAS, do I need to install it like an entire python package locally?

For example, something like this tutorial Build Python Packages Without Publishing

Define tags for View routes

How to define tags for PydanticViews? The usual solution with @docs annotation does not work:

from aiohttp import web
from aiohttp_apispec import docs
from aiohttp_pydantic import PydanticView

from my_models import SomeModel

routes = web.RouteTableDef()


@routes.view('/pipeline')
class PipelineView(PydanticView):
    @docs(tags=['post_tag'])
    async def post(self, some_pydantic_model: SomeModel) -> r201:
        # business logic here

        return web.json_response(
            {'answer': 42},
            status=web.HTTPCreated.status_code)

Integration with Dependency Injector

There is an issue to use aiohttp-pydantic with (dependency-injector).
The issue is in pydantic exception about DI params, which will be injected.
In order to fix it, we need to override PydanticView logic in static method parse_func_signature (actually in injectors module, _parse_func_signature). We need to have a mechanism to skip DI and our custom classes/types in order to allow DI inject services/resources to aiohttp handlers.

There is a fix with defining our custom 'PydanticView'-like class, but in this case we need to override almost all oas package too, because there is a call of is_pydantic_view function to build spec in generate_oas.

Broken Compat with Pydantic latest

The Group parameters are broken with the latest pydantic version.

from __future__ import annotations

import logging
from aiohttp import web
from aiohttp_pydantic import PydanticView
from aiohttp_pydantic.injectors import Group

logging.basicConfig(level=logging.DEBUG)


class Test(Group):
    name: str
    age: int = 123


class TestView(PydanticView):
    async def get(self, test: Test):
        return web.Response(text=test.name)


app = web.Application()
app.router.add_view("/test", TestView)
web.run_app(app)

Query to http://0.0.0.0:8080/test?name=foo returns

[{"loc": ["test"], "msg": "field required", "type": "value_error.missing", "in": "query string"}]

Group and Validators

Group does not respect @validators.

Response model validation

class PetCollectionView(PydanticView):
    async def get(self, age: Optional[int] = None) -> r200[List[Pet]]:
         pass

I expected this handler's response annotation triggers response content validation with pydantic, but unfortunately I was wrong. Is there any library capability to do that just like FastAPI does? It would very frustrating to have only request validation within aiohttp server, but without the same approach (sugar) for response validation

Uncaptured query params

Hello!
I have got an API with multiple models. I'd like to to implement filters in my handler which would look like this:
GET /api/pets?name.eq=buddy&age.lt=5. Here I'm trying to get all pets named buddy younger than 5 years. There are two problems: my parameter names are not valid python identifiers(yet it could be solved by using _ delimiter). If first problem is solved, the second is lots of combinations to field_name.filter_type to list and they should be listed for every model I would like to filter.
It would be more convenient for me to get all query parameters not captured by keyword arguments, to handle it by myself later. Like this

class Pet(PydanticView):
    def get(self, **filters: str):
        pets = PetModel.filter(filters)
        ....

But It turns out there's no way to express it in my handler signature right now.

In swagger json it looks like this:

{
  "paths": {
    "/api/pets": {
      "get": {
        "parameters": [
          {
            "in": "query",
            "name": "filter",
            "schema": {
              "$ref": "#/components/schemas/FilterQuery"
            },
            "style": "form",
            "explode": true,
            "description": "Filter fields by eq(==), ne(!=), lt(<), le(<=), gt(>), ge(>=) methods",
            "required": false
          }
        ]
      }
    }
  },
  "components": {
    "schemas": {
      "FilterQuery": {
        "type": "object",
        "properties": {
          "param": {
            "type": "string",
            "default": "value"
          }
        },
        "example": {
          "name": "string",
          "id.ge": 1,
          "id.ne": 2,
          "created_at.gt": "2012-12-12T12:12:12+00:00",
          "updated_at.lt": "2021-12-12T12:12:12+00:00"
        }
      }
    }
  }
}

Creating link references

What are the plans for representing links?
https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.1.md#linkObject

With some input for alignment.. I'm happy to contribute code and PR.

Reloading the Swagger UI page breaks it

Steps to reproduce:

run the demo
go to http://127.0.0.1:8080/oas and reload the page
click on the POST endpoint

Observed behavior:

The steps above lead to the following error: Could not resolve reference: Could not resolve pointer: /components/schemas/Friend does not exist in document
Nested schemas, such as Friend in the demo, can not be loaded after the refresh. The outer schemas (Pet in the demo) are loaded fine. I observe the same behavior in my own project using aiohttp-pydantic.

Tested with Python 3.8.5. Emptying the browser cache does not help.

export a handle style decorator

I'm using handle (async def handle(request) -> Response) in my aiohttp application.

it looks like currently it only support class based view, not request.

I try a code snippet to make it work:
(most of code are copied from inner)

from asyncio import iscoroutinefunction
from functools import update_wrapper
from typing import Callable, Iterable

from aiohttp import web
from aiohttp.web_response import json_response, StreamResponse
from aiohttp_pydantic.injectors import (
    CONTEXT, AbstractInjector, _parse_func_signature, MatchInfoGetter, BodyGetter, QueryGetter, HeadersGetter,
)
from pydantic import ValidationError


async def on_validation_error(exception: ValidationError, context: CONTEXT) -> StreamResponse:
    """
    This method is a hook to intercept ValidationError.

    This hook can be redefined to return a custom HTTP response error.
    The exception is a pydantic.ValidationError and the context is "body",
    "headers", "path" or "query string"
    """
    errors = exception.errors()
    for error in errors:
        error["in"] = context

    return json_response(data=errors, status=400)


def parse_func_signature(func: Callable) -> Iterable[AbstractInjector]:
    path_args, body_args, qs_args, header_args, defaults = _parse_func_signature(func)
    injectors = []

    def default_value(args: dict) -> dict:
        """
        Returns the default values of args.
        """
        return {name: defaults[name] for name in args if name in defaults}

    if path_args:
        injectors.append(MatchInfoGetter(path_args, default_value(path_args)))
    if body_args:
        injectors.append(BodyGetter(body_args, default_value(body_args)))
    if qs_args:
        injectors.append(QueryGetter(qs_args, default_value(qs_args)))
    if header_args:
        injectors.append(HeadersGetter(header_args, default_value(header_args)))
    return injectors


def decorator(handler):
    """
    Decorator to unpack the query string, route path, body and http header in
    the parameters of the web handler regarding annotations.
    """

    injectors = parse_func_signature(handler)

    async def wrapped_handler(request):
        args = []
        kwargs = {}
        for injector in injectors:
            try:
                if iscoroutinefunction(injector.inject):
                    await injector.inject(request, args, kwargs)
                else:
                    injector.inject(request, args, kwargs)
            except ValidationError as error:
                return await on_validation_error(error, injector.context)

        return await handler(*args, **kwargs)

    update_wrapper(wrapped_handler, handler)
    return wrapped_handler


@decorator
async def get(id: int = None, /, with_comments: bool = False, *, user_agent: str = None):
    return web.json_response(
        {
            'id': id,
            'UA': user_agent,
            'with_comments': with_comments,
        }
    )


app = web.Application()
app.router.add_get('/{id}', get)

if __name__ == '__main__':
    web.run_app(app, port=9092)

there is another sugestion:

all *Getter.inject can be asynchronous function, and just call await injector.inject(request, args, kwargs) without iscoroutinefunction call. ( and put BodyGetter at last )

Complex types in Groups

I am trying to use Groups for multiple query parameters.

class Filters(Group):
    author: Optional[str]
    state: Optional[State]
    game: Optional[Game]
    talent: Optional[Talent]

But I am getting an error

    class View(PydanticView):
../../../opt/miniconda3/envs/fcast-backend/lib/python3.9/abc.py:106: in __new__
    cls = super().__new__(mcls, name, bases, namespace, **kwargs)
../../../opt/miniconda3/envs/fcast-backend/lib/python3.9/site-packages/aiohttp_pydantic/view.py:56: in __init_subclass__
    decorated_handler = inject_params(handler, cls.parse_func_signature)
../../../opt/miniconda3/envs/fcast-backend/lib/python3.9/site-packages/aiohttp_pydantic/view.py:118: in inject_params
    injectors = parse_func_signature(handler)
../../../opt/miniconda3/envs/fcast-backend/lib/python3.9/site-packages/aiohttp_pydantic/view.py:88: in parse_func_signature
    injectors.append(QueryGetter(qs_args, default_value(qs_args)))
../../../opt/miniconda3/envs/fcast-backend/lib/python3.9/site-packages/aiohttp_pydantic/injectors.py:113: in __init__
    self.model = type("QueryModel", (BaseModel,), attrs)
pydantic/main.py:198: in pydantic.main.ModelMetaclass.__new__
    ???
pydantic/fields.py:506: in pydantic.fields.ModelField.infer
    ???
pydantic/fields.py:436: in pydantic.fields.ModelField.__init__
    ???
pydantic/fields.py:557: in pydantic.fields.ModelField.prepare
    ???
pydantic/fields.py:831: in pydantic.fields.ModelField.populate_validators
    ???
pydantic/validators.py:765: in find_validators
    ???
E   RuntimeError: no validator found for <class 'app.schemas.vacancy.VacancyFilters'>, see `arbitrary_types_allowed` in Config

Is there a way to use complex types. In my case those types are Enums or it's supposed to work only with primitive types?

BUG: UI issues using View without request params

oas setup doesn't work right when view method doesn't have extra params. It's not unusual for a get request to have no request data.

This example doesn't work:

from typing import Optional

from aiohttp import web
from aiohttp_pydantic import PydanticView, oas
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):
        return web.json_response({'name': 'foobar'})


app = web.Application()
oas.setup(app)
app.router.add_view('/article', ArticleView)
web.run_app(app)

Here is what happens in the UI

400 instead of 422

When pydantic validation failed (post request), automatic 400 raise instead 422.
How can we include 422 default sheme validation in oas or other user schemas?!

Optional body parameters

The README describes how query parameters are optional and how default values can be set.

It might be up for debate, whether this is a good idea, but here it goes:

It would be nice, if it were possible to declare body parameters as optional, too, and provide default values.

If you don't like this idea, it would be nice on the other hand, if there were an error.

Currently, if one tries this, one only gets the default values in a dictionary; independent of what the request body contains.

To illustrate this situation, I have prepared a little script:
https://gist.github.com/crazyscientist/2358f59bed61273c82ef583abb64e2f9

pydantic type

I am trying to follow this tutorial for recursive pydantic models.

I am getting tripped up on the class ReadMultModel that calls class ReadSingleModel

Is this legit Pydantic?

class ReadMultModel(BaseModel):
    devices: Dict[str, ReadSingleModel]

This is my models.py below and

from typing import Any, AsyncIterator, Awaitable, Callable, Dict
from pydantic import BaseModel


class ReadSingleModel(BaseModel):
    address: str
    object_type: str
    object_instance: str

    
    
class ReadMultModel(BaseModel):
    devices: Dict[str, ReadSingleModel]

In Insomnia I am trying to do this:
192.168.0.105:8080/bacnet/read/multiple


{"devices":{
    "boiler":{
    "address":"12345:2",
    "object_type":"analogInput",
    "object_instance":"2"
    },
    "cooling_plant":{
    "address":"12345:2",
    "object_type":"analogInput",
    "object_instance":"2"
    },
    "air_handler_1":{
    "address":"12345:2",
    "object_type":"analogInput",
    "object_instance":"2"
    },
    "air_handler_2":{
    "address":"12345:2",
    "object_type":"analogInput",
    "object_instance":"2"
    },
    "hot_water_valve_1":{
    "address":"12345:2",
    "object_type":"analogInput",
    "object_instance":"2"
    }
}}

This will error: AttributeError: 'ReadMultModel' object has no attribute 'address'

This is main.py for what its worth:

from aiohttp.web import Application, json_response, middleware
import asyncio
from pathlib import Path
from aiohttp_pydantic import PydanticView
from aiohttp import web
from aiohttp_pydantic import oas
from models import ReadSingleModel,WriteSingleModel,ReleaseSingleModel
from models import ReadMultModel


app = Application()
oas.setup(app, version_spec="1.0.1", title_spec="BACnet Rest API App")


# Create your PydanticView and add annotations.
class ReadSingleView(PydanticView):
    async def get(self, bacnet_req: ReadSingleModel):
        read_result = [
        bacnet_req.address,
        bacnet_req.object_type,
        bacnet_req.object_instance
        ]
        response_obj = {"status":"success", "present_value" : read_result}
        return web.json_response(response_obj)


class ReadMultView(PydanticView):
    async def get(self, bacnet_req: ReadMultModel):
        for device,values in bacnet_req:
        
            read_result = [
            bacnet_req.address,
            bacnet_req.object_type,
            bacnet_req.object_instance
            ]
            
            device_mapping[device] = {'pv':read_result_round}

        response_obj = {"status":"success", "data": device_mapping }    
        return web.json_response(response_obj)


app.router.add_view('/bacnet/read/single', ReadSingleView)
app.router.add_view('/bacnet/read/multiple', ReadMultView)
web.run_app(app, host='0.0.0.0', port=8080)

It would be nice to be able to use an input model for GET requests too.

This is a feature request, but it'd be nice to be able to use an input model as opposed to e.g. 10 parameters for GET requests.

Duplicated query parameters

Hello!

Encountered bug in PydanticView.

Considering this simple view with single query parameter:

class SimpleView(PydanticView):
    async def get(self, foo: str):
        ...

Request with more than one query parameter foo raises a TypeError
GET /path_to_simple_view?foo=123&foo=456

Traceback (most recent call last):
  File "/Users/vadim.pochivalin/Projects/COREMSA/goods/goods-api/app/utils.py", line 44, in catch_exceptions
    resp = await handler(request)
  File "/Users/vadim.pochivalin/Projects/COREMSA/goods/goods-api/venv/lib/python3.8/site-packages/aiohttp_pydantic/view.py", line 23, in _iter
    resp = await method()
  File "/Users/vadim.pochivalin/Projects/COREMSA/goods/goods-api/venv/lib/python3.8/site-packages/aiohttp_pydantic/view.py", line 87, in wrapped_handler
    injector.inject(self.request, args, kwargs)
  File "/Users/vadim.pochivalin/Projects/COREMSA/goods/goods-api/venv/lib/python3.8/site-packages/aiohttp_pydantic/injectors.py", line 89, in inject
    kwargs_view.update(self.model(**request.query).dict())
TypeError: ModelMetaclass object got multiple values for keyword argument 'foo'

Error: Malformed JSON

I get {"error": "Malformed JSON"} when send POST request at example code:

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, host='127.0.0.1', port=8000)

GET requests documentation issues

Hello!

There are two issues with generating open api specification for get requests:

1) While parsing group signature, in the _get_group_signature function you decide that param is optional by default value. But it's a lit incorrect because if I write in this way:

class QueryParams(Group):
    some_param: Optional[int] = None

it will be still required in the swagger:
in _add_http_method_to_oas:

 if name in defaults:
     attrs["__root__"] = defaults[name]
     oas_operation.parameters[i].required = False
 else:
     oas_operation.parameters[i].required = True

So if name is not in defaults, then parameter will be required.

in _get_group_signature:

for attr_name, type_ in base.__annotations__.items():
    if (default := attrs.get(attr_name)) is None:
        defaults.pop(attr_name, None)
    else:
        defaults[attr_name] = default

If default value is None then it value will be removed from defaults even if param type marked as Optional.

I suggest following solution:
in _get_group_signature:

for attr_name, type_ in base.__annotations__.items():
    _is_optional = getattr(type_, '_name', '') == 'Optional'
    if (default := attrs.get(attr_name)) is None and not _is_optional:
        defaults.pop(attr_name, None)
     else:
         defaults[attr_name] = default

Here I make a new variable _is_optional and decide that field is not required if it's marked as Optional.

2) The second issue: when you specify query param as any enum type, then enum model schema doesn't being added to open api schema as for body or response schema.

To fix it, you need to modify _add_http_method_to_oas function:

query_param_schema = type(name, (BaseModel,), attrs).schema(
    ref_template="#/components/schemas/{model}"
)
if def_sub_schemas := query_param_schema.pop("definitions", None):
    oas.components.schemas.update(def_sub_schemas)
oas_operation.parameters[i].schema = query_param_schema

Above I extracted sub_schema from param schema and set it to common components schema.

Could you please fix it? If not, can I fix it myself and make a PR?

Invalid specification is generated

The documentation says:

.. code-block:: python3

from aiohttp import web
from aiohttp_pydantic import oas


app = web.Application()
oas.setup(app)

But when you contact the address http://127.0.0.1:8080/oas/spec
Returns a specification that is not valid for swagger

Here's the mistake

Structural error at 
should have required property 'info' missingProperty: info
Jump to line 0

In the list, you need to return the "Info" attribute

newbie help

Hi,

Would there be any chance I could get a tip getting started? Very cool creation btw...

So I am trying the Inject Path Parameters type of API from the link in the README. And I get a {"error": "Malformed JSON"} response just trying to test out some concept code for the API.

from typing import Optional, Literal

from aiohttp import web
from aiohttp_pydantic import PydanticView
from pydantic import BaseModel



ACTION_TYPE_MAPPING = Literal["read", "write", "release"]

PRIORITY_MAPPING = Literal["1", "2", "3","4",
                           "5", "6", "7", "8",
                           "9", "10", "11", "12",
                           "13", "14", "15", "16"
                           ]

OBJECT_TYPE_MAPPING = Literal["multiStateValue", "multiStateInput", "multiStateOutput",
                       "analogValue", "analogInput", "analogOutput",
                       "binaryValue", "binaryInput", "binaryOutput"]

BOOLEAN_ACTION_MAPPING = Literal["active", "inactive"]



class ValueModel(BaseModel):
    multiStateValue: Optional[int]
    multiStateInput: Optional[int]
    multiStateOutput: Optional[int]
    analogValue: Optional[int]
    analogInput: Optional[int]
    analogOutput: Optional[int]
    binaryValue: Optional[BOOLEAN_ACTION_MAPPING]
    binaryInput: Optional[BOOLEAN_ACTION_MAPPING]
    binaryOutput: Optional[BOOLEAN_ACTION_MAPPING]


# 3 required params
class ReadRequestModel(BaseModel):
    address: str
    object_type: OBJECT_TYPE_MAPPING
    object_instance: int

# 5 required params
class WriteRequestModel(BaseModel):
    address: str
    object_type: OBJECT_TYPE_MAPPING
    object_instance: int
    value: ValueModel  # can I call the ValueModel class?
    priority: PRIORITY_MAPPING


# 4 required params
class ReleaseRequestModel(BaseModel):
    address: str
    object_type: OBJECT_TYPE_MAPPING
    object_instance: int
    priority: PRIORITY_MAPPING


class BacnetReadView(PydanticView):
    async def get(self, read_req_string: ReadRequestModel):
        print(read_req_string.split())
        return web.json_response(read_req_string)
    
class BacnetReleaseView(PydanticView):
    async def get(self, release_req_string: WriteRequestModel):
        print(release_req_string.split())
        return web.json_response(release_req_string)

class BacnetWriteView(PydanticView):
    async def get(self, write_req_string: ReleaseRequestModel):
        print(write_req_string.split())
        return web.json_response(write_req_string)


app = web.Application()
app.router.add_get('/bacnet/read/{read_req_string}', BacnetReadView)
app.router.add_get('/bacnet/write/{write_req_string}', BacnetWriteView)
app.router.add_get('/bacnet/release/{release_req_string}', BacnetReleaseView)
web.run_app(app)

In the browser when running the app, am testing this:

bacnet/read/10.200.200.27 binaryOutput 3
bacnet/write/10.200.200.27 binaryOutput 3 active 12
bacnet/release/10.200.200.27 binaryOutput 3 12

Any tips or time to read/respond greatly appreciated. I know I have a ton of issues here particularly how to parse the injection string to the URL?

Pure List Body

Hi there,

I'm testing your library, but can't figure out how declare that the body of the put or post request will be a List. I mean no json wrapper around, just a list. [{"a":"b", ...}].
I have tried in the function signature to declare like post(self, query_para: str, List[ListBody]), being ListBody the Class of objects the List will contain. The oas builder think it is a query parameter as well so throws an error that such query param is not existent.

If i do it with pydantic, I'm forced to put a tag on a new class like:
class RecordsList(BaseModel)
records: List[ListBody]
which is exactly what I need to avoid.

Am I doing something wrong, or is it a requirement for the library to do it via Pydantic ?

Thanks in advance.

call a POST request in the code

Hello!

I'm in code trying to handle an error that occurs due to invalid input data, and I'm trying to call the POST request again in on_validation_error, but I'm getting an error that the POST request is not expecting parameters

async def on_validation_error(
            self, exception: ValidationError, context: str):
        ...
        request_not_error_model = await self.post(request_input, config_models=not_error_config)
        ...

@staticmethod
async def post(request_input: AggregatorInput, config_models: dict = None):
        ...

I'm getting an error

request_not_error_model = await self.post(request_input, config_models=not_error_config)\nTypeError: AggregatorView.post() got an unexpected keyword argument 'config_models'"}

Set 'title' for model field

I have a model:

from aiohttp_pydantic import PydanticView, oas
from pydantic import BaseModel

class Task(BaseModel):
    id: str
    type: str
    on_success: Optional[str] # ID for task to run on success
    on_error: Optional[str] # ID for task to run on failure


@routes.view('/task')
class TaskView(PydanticView):
    async def post(self, task: Task):
        // some task logic

And in a Swagger UI (OAS) I get following model description:

id*: string
    title: Id
type*: string
    title: Type
on_success: string
    title: On Success
on_error: string
    title: On Error

What I want is to somehow define the title in pydantic model so that it's not plainly repeat the field itself. How do I do that?

maillol / aiohttp-pydantic Goto Github PK

aiohttp-pydantic's People

Contributors

Stargazers

Watchers

Forkers

aiohttp-pydantic's Issues

Steps to reproduce:

Observed behavior:

1) While parsing group signature, in the _get_group_signature function you decide that param is optional by default value. But it's a lit incorrect because if I write in this way:

2) The second issue: when you specify query param as any enum type, then enum model schema doesn't being added to open api schema as for body or response schema.

Recommend Projects

Recommend Topics

Recommend Org