Json api service on aiohttp: middleware and validation

In this article I will describe one of the approaches for creating a json api service with data validation.

The service will be implemented on aiohttp . It is a modern, constantly evolving python framework that uses asyncio



.

About annotations:

The introduction of annotations in the python



code made it easier to understand. Also, annotations open up some additional possibilities. It is annotations that play a key role in data validation for api method handlers in this article.

Libraries used:

• aiohttp - a framework for building web applications
• pydantic - classes that allow you to declaratively describe data and validate it
• valdec - decorator for validating arguments and return values ​​of functions

Table of contents:

• 1. Files and folders of the application
• 2.json middlewares
  
  
  • 2.1. Simple middleware for json service
    
    
    • 2.1.1. Handler declaration
    • 2.1.2. SimpleHandler class for middleware
      
      
      • 2.1.2.1. Middleware method
      • 2.1.2.2. Method for retrieving error response data
      • 2.1.2.3. Handler start method
    • 2.1.3. Examples of
      
      
      • 2.1.3.1. Reply with code 200
      • 2.1.3.2. Reply with code 400
      • 2.1.3.3. Reply with code 500
  • 2.2. middleware for "kwargs handlers"
    
    
    • 2.2.1. Handler declaration
    • 2.2.2. ArgumentsManager helper class
    • 2.2.3. KwargsHandler class for middleware
      
      
      • 2.2.3.1. Handler start method
      • 2.2.3.2.
    • 2.2.4.
      
      
      • 2.2.4.1. /create
      • 2.2.4.2. /read
      • 2.2.4.3. /info/{info_id}
  • 2.3. middleware c /
    
    
    • 2.3.1. pydantic.BaseModel
      
      
      • 2.3.1.1.
      • 2.3.1.2.
    • 2.3.2. valdec.validate
      
      
      • 2.3.2.1. -
      • 2.3.2.2.
      • 2.3.2.3.
      • 2.3.2.4.
    • 2.3.3.
    • 2.3.4.
    • 2.3.5.
    • 2.3.6. WrapsKwargsHandler middleware
      
      
      • 2.3.6.1.
      • 2.3.6.2.
    • 2.3.7.
      
      
      • 2.3.7.1. /create
      • 2.3.7.2. /read
      • 2.3.7.3. /info/{info_id}
• 3.
• 4.

1.

- sources -    
    - data_classes -     
        - base.py -   
        - person.py -    
        - wraps.py -     /
    - handlers -     
        - kwargs.py -      `KwargsHandler.middleware`
        - simple.py -      `SimpleHandler.middleware`
        - wraps.py -      `WrapsKwargsHandler.middleware`
    - middlewares -     middlewares
        - exceptions.py -  
        - kwargs_handler.py -  `KwargsHandler`
        - simple_handler.py -  `SimpleHandler`
        - utils.py -      middlewares
        - wraps_handler.py -  `WrapsKwargsHandler`
    - requirements.txt -  
    - run_kwargs.py -   `KwargsHandler.middleware`
    - run_simple.py -  c `SimpleHandler.middleware`
    - run_wraps.py -  c `WrapsKwargsHandler.middleware`
    - settings.py -    
- Dockerfile -    
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

: https://github.com/EvgeniyBurdin/api_service

2. json middlewares

middleware



aiohttp.web.Application()



.

middleware



, , . . middleware



.

middleware



, .

middleware



"" "" web.Request



web.Response



. .

, middleware



/, .

, .

2.1. middleware json

, aiohttp.web.Application()



, , :

from aiohttp import web

async def some_handler(request: web.Request) -> web.Response:
    data = await request.json()
    ...
    text = json.dumps(some_data)
    ...
    return web.Response(text=text, ...)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

"" web.Request



, json. , . json "" web.Response



( web.json_response()



).

2.1.1.

. , middleware



, , :

from aiohttp import web

async def some_handler(request: web.Request, data: Any) -> Any:
    ...
    return some_data
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

. web.Request



( ), — python, .

, : data: Any



. ( ), , "" . .

, , :

from aiohttp import web
from typing import Union, List

async def some_handler(
    request: web.Request, data: Union[str, List[str]]
) -> List[int]:
    ...
    return some_data
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

2.1.2. SimpleHandler



middleware

SimpleHandler



middleware



, / middleware



( ).

.

2.1.2.1. middleware

    @web.middleware
    async def middleware(self, request: web.Request, handler: Callable):
        """ middleware  json-.
        """
        if not self.is_json_service_handler(request, handler):
            return await handler(request)

        try:
            request_body = await self.get_request_body(request, handler)

        except Exception as error:
            response_body = self.get_error_body(request, error)
            status = 400

        else:
            #  
            response_body, status = await self.get_response_body_and_status(
                request, handler, request_body
            )

        finally:
            #     python (  
            # response_body)   json.
            text, status = await self.get_response_text_and_status(
                request, response_body, status
            )

        return web.Response(
            text=text, status=status, content_type="application/json",
        )
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

middlewares



.

, :

    ...
    app = web.Application()
    service_handler = SimpleHandler()
    app.middlewares.append(service_handler.middleware)
    ...
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

2.1.2.2.

json , , , ( 400), ( 500), json.

"" :

    def get_error_body(self, request: web.Request, error: Exception) -> dict:
        """       .
        """
        return {"error_type": str(type(error)), "error_message": str(error)}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , json. , json .

2.1.2.3.

:

    async def run_handler(
        self, request: web.Request, handler: Callable, request_body: Any
    ) -> Any:
        """   ,     .
        """
        return await handler(request, request_body)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, / .

2.1.3.

:

async def some_handler(request: web.Request, data: dict) -> dict:
    return data
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

url .

{
    "name": "test",
    "age": 25
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "name": "test",
    "age": 25
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "name": "test", 111111111111
    "age": 25
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "error_type": "<class 'json.decoder.JSONDecodeError'>",
    "error_message": "Expecting property name enclosed in double quotes: line 2 column 21 (char 22)"
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

async def handler500(request: web.Request, data: dict) -> dict:
    raise Exception("  500")
    return data
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "name": "test",
    "age": 25
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "error_type": "<class 'Exception'>",
    "error_message": "  500"
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

2.2. middleware "kwargs-"

middleware



.

.

:

async def some_handler(request: web.Request, data: dict) -> dict:

    storage = request.app["storage"]
    logger = request.app["logger"]
    user_id = request.match_info["user_id"]
    #  ..  ....

    return data
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

storage



, logger



( - ), , "" .

2.2.1.

, , , :

async def some_handler_1(data: dict) -> int:
    # ...
    return some_data

async def some_handler_2(storage: StorageClass, data: List[int]) -> dict:
    # ...
    return some_data

async def some_handler_3(
    data: Union[dict, List[str]], logger: LoggerClass, request: web.Request
) -> str:
    # ...
    return some_data
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, .

2.2.2. ArgumentsManager

middleware



, "" "" .

, "" ArgumentsManager



. middlewares/utils.py



( ).

" " — " " , " ", — , " ".

, :

@dataclass
class RawDataForArgument:

    request: web.Request
    request_body: Any
    arg_name: Optional[str] = None

class ArgumentsManager:
    """    .

            ,    
          .
    """

    def __init__(self) -> None:

        self.getters: Dict[str, Callable] = {}

    #  json  ------------------------------------------------------

    def reg_request_body(self, arg_name) -> None:
        """      .
        """
        self.getters[arg_name] = self.get_request_body

    def get_request_body(self, raw_data: RawDataForArgument):
        return raw_data.request_body

    #   request --------------------------------------------------------

    def reg_request_key(self, arg_name) -> None:
        """       request.
        """
        self.getters[arg_name] = self.get_request_key

    def get_request_key(self, raw_data: RawDataForArgument):
        return raw_data.request[raw_data.arg_name]

    #   request.app ----------------------------------------------------

    def reg_app_key(self, arg_name) -> None:
        """       app.
        """
        self.getters[arg_name] = self.get_app_key

    def get_app_key(self, raw_data: RawDataForArgument):
        return raw_data.request.app[raw_data.arg_name]

    #   ------------------------------------------------------

    def reg_match_info_key(self, arg_name) -> None:
        """        .
        """
        self.getters[arg_name] = self.get_match_info_key

    def get_match_info_key(self, raw_data: RawDataForArgument):
        return raw_data.request.match_info[raw_data.arg_name]

    #     ...
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

web.Application()



:

# ...

app = web.Application()

arguments_manager = ArgumentsManager()

#    ,    
#    json- 
arguments_manager.reg_request_body("data")

#    ,    
#      request.match_info
arguments_manager.reg_match_info_key("info_id")

#     
# (  "  "     )
app["storage"] = SomeStorageClass(login="user", password="123")
#    ,    
#  
arguments_manager.reg_app_key("storage")

# ...
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

ArgumentsManager



. middleware



:

...
service_handler = KwargsHandler(arguments_manager=arguments_manager)
app.middlewares.append(service_handler.middleware)
...
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

. , , … , , .

2.2.3. KwargsHandler



middleware

KwargsHandler



SimpleHandler



, .2.2.1.

— run_handler



, — make_handler_kwargs



build_error_message_for_invalid_handler_argument



( ).

2.2.3.1.

:

    async def run_handler(
        self, request: web.Request, handler: Callable, request_body: Any
    ) -> Any:
        """   ,     .

            (   ,   
             //)
        """
        kwargs = self.make_handler_kwargs(request, handler, request_body)

        return await handler(**kwargs)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, . , . .

2.2.3.2.

make_handler_kwargs



. , . ArgumentsManager



.

, , ArgumentsManager



.

. , web.Request



, web.Request



(, r: web.Request



req: web.Request



request: web.Request



). , web.Request



"" , .

: .

build_error_message_for_invalid_handler_argument



— . .

2.2.4.

:

async def create(
    data: Union[dict, List[dict]], storage: dict,
) -> Union[dict, List[dict]]:
    # ...

async def read(storage: dict, data: str) -> dict:
    # ...

async def info(info_id: int, request: web.Request) -> str:
    # ...
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

POST



, — GET



(, )

[
    {
        "name": "Ivan"
    },
    {
        "name": "Oleg"
    }
]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

[
    {
        "id": "5730bab1-9c1b-4b01-9979-9ad640ea5fc1",
        "name": "Ivan"
    },
    {
        "id": "976d821a-e871-41b4-b5a2-2875795d6166",
        "name": "Oleg"
    }
]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

"5730bab1-9c1b-4b01-9979-9ad640ea5fc1"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "id": "5730bab1-9c1b-4b01-9979-9ad640ea5fc1",
    "name": "Ivan"
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

"any json"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

"info_id=123 and request=<Request GET /info/123 >"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

2.3. middleware c /

, api- .

, create



:

{
    "data": [
        {
            "name": "Ivan"
        },
        {
            "name": "Oleg"
        }
    ],
    "id": 11
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

:

{

    "success": true,
    "result": [
        {
            "id": "9738d8b8-69da-40b2-8811-b33652f92f1d",
            "name": "Ivan"
        },
        {
            "id": "df0fdd43-4adc-43cd-ac17-66534529d440",
            "name": "Oleg"
        }
    ],
    "id": 11
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, data



result



.

id



, .

success



.

, :

read



:

{
    "data":  "ddb0f2b1-0179-44b7-b94d-eb2f3b69292d",
    "id": 3
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

:

{
    "success": false,
    "result": {
        "error_type": "<class 'handlers.PersonNotFound'>",
        "error_message": "Person whith id=ddb0f2b1-0179-44b7-b94d-eb2f3b69292d not found!"
    },
    "id": 3
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

json middleware



middleware



. run_handler



, ( ) get_error_body



.

, "" , ( data



). ( result



). middleware



.

, , .

" ", . .

2.3.1. pydantic.BaseModel

pydantic.BaseModel



.

( ). — .

:

from pydantic import BaseModel
from typing import Union, List

class Info(BaseModel):
    foo: int

class Person(BaseModel):
    name: str
    info: Union[Info, List[Info]]

kwargs = {"name": "Ivan", "info": {"foo": 0}}
person = Person(**kwargs)
assert person.info.foo == 0

kwargs = {"name": "Ivan", "info": [{"foo": 0}, {"foo": 1}]}
person = Person(**kwargs)
assert person.info[1].foo == 1

kwargs = {"name": "Ivan", "info": {"foo": "bar"}}  # <- , str  int
person = Person(**kwargs)
#  :
# ...
# pydantic.error_wrappers.ValidationError: 2 validation errors for Person
# info -> foo
#  value is not a valid integer (type=type_error.integer)
# info
#  value is not a valid list (type=type_error.list)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , . , , .

typing



.

- pydantic.BaseModel



, "" ( … , "" — ).

. , : info.foo



int



, info



list



, .

pydantic.BaseModel



, .

2.3.1.1.

, , :

kwargs = {"name": "Ivan", "info": {"foo": "0"}}
person = Person(**kwargs)
assert person.info.foo == 0
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, . str->int



( pydantic



. ).

, , , UUID



-> UUID



. , , , Strict...



. , pydantic.StrictInt



, pydantic.StrictStr



, ....

2.3.1.2.

, , :

kwargs = {"name": "Ivan", "info": {"foo": 0}, "bar": "BAR"}
person = Person(**kwargs)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

.

, .

, , :

from pydantic import BaseModel, Extra, StrictInt, StrictStr
from typing import Union, List

class BaseApi(BaseModel):
    class Config:
        #    (ignore),  (allow) 
        #  (forbid)     
        # , :
        # https://pydantic-docs.helpmanual.io/usage/model_config/
        extra = Extra.forbid

class Info(BaseApi):
    foo: StrictInt

class Person(BaseApi):
    name: StrictStr
    info: Union[Info, List[Info]]

kwargs = {"name": "Ivan", "info": {"foo": 0}, "bar": "BAR"}
person = Person(**kwargs)
# ...
# pydantic.error_wrappers.ValidationError: 1 validation error for Person
# bar
#   extra fields not permitted (type=value_error.extra)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

— , .

2.3.2. valdec.validate

valdec.validate / .

, .

, None



( -> None:



).

/:

from valdec.decorators import validate

@validate  #     ,  
def foo(i: int, s: str) -> int:
    return i

@validate("i", "s")  #   "i"  "s"
def bar(i: int, s: str) -> int:
    return i
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

… .

#   
from valdec.decorators import async_validate as validate

@validate("s", "return", exclude=True)  #   "i"
async def foo(i: int, s: str) -> int:
    return int(i)

@validate("return")  #   
async def bar(i: int, s: str) -> int:
    return int(i)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

2.3.2.1. -

/ , - ( , ), , , .

-:

def validator(
    annotations: Dict[str, Any],
    values: Dict[str, Any],
    is_replace: bool,
    extra: dict
) -> Optional[Dict[str, Any]]:
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

:

• annotations
  
  
  
  — , .
• values
  
  
  
  — , .
• is_replace
  
  
  
  — , -, — .
  
  
  • True
    
    
    
    , . , , BaseModel
    
    
    
    , BaseModel
    
    
    
    , " ".
  • False
    
    
    
    , None
    
    
    
    , ( , , , BaseModel
    
    
    
    ).
• extra
  
  
  
  — .

, validate



- pydantic.BaseModel



.

:

• ( pydantic.BaseModel
  
  
  
  )
• . .
• ( ), , is_replace
  
  
  
  .

, , , . , , , .

- ( valdec



ValidatedDC



). : , pydantic.BaseModel



. , , "" .

2.3.2.2.

, "" :

from typing import List, Optional

from pydantic import BaseModel, StrictInt, StrictStr
from valdec.decorators import validate

class Profile(BaseModel):
    age: StrictInt
    city: StrictStr

class Student(BaseModel):
    name: StrictStr
    profile: Profile

@validate("group")
def func(group: Optional[List[Student]] = None):
    for student in group:
        assert isinstance(student, Student)
        assert isinstance(student.name, str)
        assert isinstance(student.profile.age, int)

data = [
    {"name": "Peter", "profile": {"age": 22, "city": "Samara"}},
    {"name": "Elena", "profile": {"age": 20, "city": "Kazan"}},
]

func(data)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

assert'



.

:

@validate  #  
def func(group: Optional[List[Student]] = None, i: int) -> List[Student]:
    #...
    return [
        {"name": "Peter", "profile": {"age": 22, "city": "Samara"}},
        {"name": "Elena", "profile": {"age": 20, "city": "Kazan"}},
    ]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , return



, Student



( ).

… . , , (, , ). :

from valdec.data_classes import Settings
from valdec.decorators import validate as _validate
from valdec.validator_pydantic import validator

custom_settings = Settings(
    validator=validator,     # -.
    is_replace_args=False,   #     
    is_replace_result=False, #     
    extra={}                 #  ,  
                             #   -
)
#   
def validate_without_replacement(*args, **kwargs):
    kwargs["settings"] = custom_settings
    return _validate(*args, **kwargs)

# 
@validate_without_replacement
def func(group: Optional[List[Student]] = None, i: int) -> List[Student]:
    #...
    return [
        {"name": "Peter", "profile": {"age": 22, "city": "Samara"}},
        {"name": "Elena", "profile": {"age": 20, "city": "Kazan"}},
    ]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

func



, is_replace_result=False



. , is_replace_args=False



.

, .

— , , "" . , . , .

— , , , , ? — .

, -, — .

2.3.2.3.

:

from valdec.decorators import validate

@validate
def foo(i: int):
    assert isinstance(i, int)

foo("1")
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

. , .

, , validate



, - pydantic.BaseModel



. .2.3.1.1. .

, ( ), :

from valdec.decorators import validate
from pydantic import StrictInt

@validate
def foo(i: StrictInt):
    pass

foo("1")
# ...
# valdec.errors.ValidationArgumentsError: Validation error
# <class 'valdec.errors.ValidationError'>: 1 validation error for
# argument with the name of:
# i
#  value is not a valid integer (type=type_error.integer).
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

: , .

.

2.3.2.4.

• valdec.errors.ValidationArgumentsError
  
  
  
  — ""
• valdec.errors.ValidationReturnError
  
  
  
  —

. pydantic.BaseModel



.

2.3.3.

, - pydantic.BaseModel



.

C :

data_classes/base.py

from pydantic import BaseModel, Extra

class BaseApi(BaseModel):
    """     api.
    """
    class Config:
        extra = Extra.forbid
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

2.3.4.

middleware, , , , :

from typing import List, Union
from valdec.decorators import async_validate as validate
from data_classes.person import PersonCreate, PersonInfo

@validate("data", "return")
async def create(
    data: Union[PersonCreate, List[PersonCreate]], storage: dict,
) -> Union[PersonInfo, List[PersonInfo]]:
    # ...
    return result
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

( ):

• validate
  
  
  
  , ""
• .

/ , .

: web.Request()



, , wep.Aplication()



. , , - , web.Request()



.

, :

data_classes/person.py

from uuid import UUID
from pydantic import Field, StrictStr
from data_classes.base import BaseApi

class PersonCreate(BaseApi):
    """    .
    """
    name: StrictStr = Field(description=".", example="Oleg")

class PersonInfo(BaseApi):
    """   .
    """
    id: UUID = Field(description=".")
    name: StrictStr = Field(description=".")
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

2.3.5.

.2.3. .

.

data_classes/wraps.py

from typing import Any, Optional
from pydantic import Field, StrictInt
from data_classes.base import BaseApi

_ID_DESCRIPTION = "   ."

class WrapRequest(BaseApi):
    """ .
    """
    data: Any = Field(description=" .", default=None)
    id: Optional[StrictInt] = Field(description=_ID_DESCRIPTION)

class WrapResponse(BaseApi):
    """ .
    """
    success: bool = Field(description=" .", default=True)
    result: Any = Field(description=" .")
    id: Optional[StrictInt] = Field(description=_ID_DESCRIPTION)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

middleware



.

2.3.6. WrapsKwargsHandler



middleware

WrapsKwargsHandler



KwargsHandler



, ( ).

— run_handler



get_error_body



.

2.3.6.1.

:

async def run_handler(
        self, request: web.Request, handler: Callable, request_body: Any
    ) -> dict:

        id_ = None

        try:
            #    
            wrap_request = WrapRequest(**request_body)

        except Exception as error:
            message = f"{type(error).__name__} - {error}"
            raise InputDataValidationError(message)

        #   id  
        id_ = wrap_request.id
        request[KEY_NAME_FOR_ID] = id_

        try:
            result = await super().run_handler(
                request, handler, wrap_request.data
            )
        except ValidationArgumentsError as error:
            message = f"{type(error).__name__} - {error}"
            raise InputDataValidationError(message)

        #    
        wrap_response = WrapResponse(success=True, result=result, id=id_)

        return wrap_response.dict()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

. InputDataValidationError



:

• ( )
• data



id
  
  
  
  
• id



StrictInt



None
  
  
  
  

id



, wrap_request.id



None



. data



. , , wrap_request.data



None



.

wrap_request.id



request



. ( ).

, wrap_request.data



(, wrap_request.data



python , json). , InputDataValidationError



valdec.errors.ValidationArgumentsError



.

, , WrapResponse



.

, . wrap_response



, ( ). , , , , , BaseApi



. , json. , "" WrapResponse.result



wrap_response



wrap_response.dict()



( ).

2.3.6.2.

:

def get_error_body(self, request: web.Request, error: Exception) -> dict:
        """         .
        """
        result = dict(error_type=str(type(error)), error_message=str(error))
        #         ,   
        #   ""
        response = dict(
            #   id    request .
            success=False, result=result, id=request.get(KEY_NAME_FOR_ID)
        )
        return response
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

( super()



result



), . .

2.3.7.

:

@validate("data", "return")
async def create(
    data: Union[PersonCreate, List[PersonCreate]], storage: dict,
) -> Union[PersonInfo, List[PersonInfo]]:
    # ...

@validate("data", "return")
async def read(storage: dict, req: web.Request, data: UUID) -> PersonInfo:
    # ...

@validate("info_id")
async def info(info_id: int, request: web.Request) -> Any:
    return f"info_id={info_id} and request={request}"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

POST , — GET (, )

{
    "data": [
        {
            "name": "Ivan"
        },
        {
            "name": "Oleg"
        }
    ],
    "id": 1
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "success": true,
    "result": [
        {
            "id": "af908a90-9157-4231-89f6-560eb6a8c4c0",
            "name": "Ivan"
        },
        {
            "id": "f7d554a0-1be9-4a65-bfc2-b89dbf70bb3c",
            "name": "Oleg"
        }
    ],
    "id": 1
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "data": {
        "name": "Eliza"
    },
    "id": 2
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "success": true,
    "result": {
        "id": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",
        "name": "Eliza"
    },
    "id": 2
}

      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "data": 123,
    "id": 3
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "success": false,
    "result": {
        "error_type": "<class 'middlewares.exceptions.InputDataValidationError'>",
        "error_message": "ValidationArgumentsError - Validation error <class 'valdec.errors.ValidationError'>: 2 validation errors for argument with the name of:\ndata\n  value is not a valid dict (type=type_error.dict)\ndata\n  value is not a valid list (type=type_error.list)."
    },
    "id": 3
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "data": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",
    "id": 4
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "success": true,
    "result": {
        "id": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",
        "name": "Eliza"
    },
    "id": 4

      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "some_key": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",
    "id": 5
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "success": false,
    "result": {
        "error_type": "<class 'middlewares.exceptions.InputDataValidationError'>",
        "error_message": "ValidationError - 1 validation error for WrapRequest\nsome_key\n  extra fields not permitted (type=value_error.extra)"
    },
    "id": null
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

{
    "success": true,
    "result": "info_id=123 and request=<Request GET /info/123 >",
    "id": null
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

3.

, WrapsKwargsHandler



, , . . pydantic.BaseModel



json-schema, ( , : swagger-, json- ).

. . , swagger



aiohttp



, ( ).

, aiohttp-swagger



( ), Union



.

aiohttp-swagger3



, , , sub_app



.

- , , , - , — .

4.

json middleware . . .

You can create any wrappers for the content of the requests and responses. Also, you can flexibly customize the validation, and apply it only where it is really needed.

I have no doubt that the examples I have proposed can be implemented in another way. But I hope that my solutions, if not completely useful, will contribute to finding other, more suitable ones.

Thank you for your time. I would be glad to receive comments and clarifications.

Used MarkConv when publishing the article