Servicio json api en aiohttp: middleware y validación

En este artículo, describiré uno de los enfoques para crear un servicio json api con validación de datos.

El servicio se implementará en aiohttp . Es un marco de Python moderno y en constante evolución que utiliza asyncio



.

Acerca de las anotaciones:

La introducción de anotaciones en el python



código facilitó la comprensión. Además, las anotaciones abren algunas posibilidades adicionales. Son las anotaciones las que juegan un papel clave en la validación de datos para los manejadores de métodos de API en este artículo.

Bibliotecas utilizadas:

• aiohttp : un marco para crear aplicaciones web
• pydantic : clases que le permiten describir datos de forma declarativa y validarlos
• valdec - decorador para validar argumentos y devolver valores de funciones

Tabla de contenido:

• 1. Archivos y carpetas de la aplicación
• 2.json middlewares
  
  
  • 2.1. Middleware simple para el servicio json
    
    
    • 2.1.1. Declaración del manejador
    • 2.1.2. Clase SimpleHandler para middleware
      
      
      • 2.1.2.1. Método de middleware
      • 2.1.2.2. Método para recuperar datos de respuesta a errores
      • 2.1.2.3. Método de inicio del controlador
    • 2.1.3. Ejemplos de
      
      
      • 2.1.3.1. Responder con código 200
      • 2.1.3.2. Responder con el código 400
      • 2.1.3.3. Responder con código 500
  • 2.2. middleware para "controladores kwargs"
    
    
    • 2.2.1. Declaración del manejador
    • 2.2.2. Clase auxiliar ArgumentsManager
    • 2.2.3. Clase KwargsHandler para middleware
      
      
      • 2.2.3.1. Método de inicio del controlador
      • 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 . . .

Puede crear envoltorios para el contenido de solicitudes y respuestas. Además, puede personalizar la validación de manera flexible y aplicarla solo donde realmente se necesite.

No tengo ninguna duda de que los ejemplos que he propuesto se pueden implementar de otra manera. Pero espero que mis soluciones, si no del todo útiles, contribuyan a encontrar otras más adecuadas.

Gracias por tu tiempo. Estaría encantado de recibir comentarios y aclaraciones.

Usó MarkConv al publicar el artículo