- 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 - докерфайл для сборки образа

middleware в aiohttp.web.Application() является оболочкой для обработчиков запросов.

Если в приложении используется middleware, то поступивший запрос сначала попадает в неё, и только потом передается в обработчик. Обработчик формирует и отдает ответ. Этот ответ снова сначала попадает в middleware и уже она отдает его наружу.

Если в приложении используются нескольно middleware, то каждая из них добавляет новый уровень вложенности.

Между middleware и обработчиком не обязательно должны передаваться "запрос" и "ответ" в виде web.Request и web.Response. Допускается передавать любые данные.

Таким образом, в middleware можно выделить действия над запросами/ответами, которые будут одинаковыми для всех обработчиков.

Это довольно упрощенное описание, но достаточное для понимания того что будет дальше.

Обычно, объявление обработчика запроса в приложении 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()).

Все обработчики нашего приложения должны выполнять подобные шаги. Поэтому, имеет смысл создать 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

    @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)
    ...

Так как у нас 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 в теле ответа.

В текущем классе он очень простой:

    async def run_handler(
        self, request: web.Request, handler: Callable, request_body: Any
    ) -> Any:
        """ Запускает реальный обработчик, и возвращает результат его работы.
        """
        return await handler(request, request_body)

Запуск выделен в отдельный метод для того, чтобы можно было добавить логику до/после выполнения самого обработчика.

Имеется такой обработчик:

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"
}

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 (или что-то еще), могут быть нужны и в других обработчиках, то везде придется "доставать" их одинаковым образом.

Хотелось бы, чтобы обработчики объявлялись, например, так:

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

То есть, чтобы нужные для обработчика сущности объявлялись в его сигнатуре и сразу были бы доступны в коде.

Про нужные для обработчика сущности должна знать middleware, чтобы она смогла "вытащить" небходимые для обработчика и "подсунуть" ему при вызове.

Переопределяется метод родительского класса:

    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)

Как можно заметить, теперь аргументы в обработчик передаются именованными. Таким образом, в обработчиках становится не важен порядок следования аргументов в сигнатуре. Но стали важны сами имена аргументов.

Метод 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 — просто формирует строку с сообщением об ошибке. Он создан для возможности изменить сообщение на свой вкус.

Сигнатуры методов такие:

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 >"

Иногда, требования для 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.

Так же, если это необходимо, можно добавить и валидацию данных.

Чтобы "два раза не вставать", я сразу покажу как добавить и оболочки и валидацию. Но сначала необходимо сделать некоторые пояснения по выбранным инструментам.

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 есть нюансы, на которые я хочу обратить внимание.

Если в любом из приведенных выше примеров заменить целое на строку, содержащую только цифры, то валидация всё равно закончится успешно:

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

Если, для определенных выше классов, попробовать выполнить такой пример:

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)

Теперь — все нормально, валидация провалилась.

Декоратор получает данные об аргументах/возврате функции, и передает их в функцию-валидатор (это сильно упрощенно, но по сути так), которая и производит, непосредственно, валидацию.

Сигнатура функции-валидатора:

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 — Словарь с дополнительными параметрами.

По умолчанию, декоратор "подменяет" исходные данные на данные экземпляра валидирующего класса:

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.

Но сама валидация данных будет работать как и раньше, не будет лишь подмены.

Есть один нюанс — линтер, иногда, может "ругаться" на различия в типах. Да, это минус. Но всегда лучше иметь выбор, чем его не иметь.

Может возникнуть вопрос — что будет, если декоратор получит, допустим, не словарь, а уже готовый экземпляр класса? Ответ — будет выполнена обычная проверка экземпляра на соответствие типу.

Как можно заметить, в настройках указывается и функция-валидатор, и если вы захотите использовать свою — именно там нужно ее подставить.

Рассмотрим такой пример применения декоратора:

from valdec.decorators import validate

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

foo("1")

Мы вызываем функцию и передаем ей строку. Но валидация прошла успешно, и в функцию прилетело целое.

• valdec.errors.ValidationArgumentsError — "поднимается" если валидация аргументов функции потерпела неудачу
• valdec.errors.ValidationReturnError — если не прошел валидацию возврат

Само сообщение с описанием ошибки берется из валидирующего класса. В нашем примере это сообщение об ошибке от pydantic.BaseModel.

Как я уже говорил, в этой статье используем классы-наследники от pydantic.BaseModel.

Cначала обязательно определим базовый класс данных:

data_classes/base.py

from pydantic import BaseModel, Extra

class BaseApi(BaseModel):
    """ Базовый класс данных для api.
    """
    class Config:
        extra = Extra.forbid

Класс для 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="Имя.")

Переопределяется метод родительского класса:

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 в варианте для успешного ответа.

Заменяется метод родительского класса:

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), но для наглядности я оставил так. Вы можете сделать как сочтете нужным.

Сигнатуры методов такие:

@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
}

У обработчиков, которые используются с классом WrapsKwargsHandler, есть всё, чтобы автоматически собрать документацию. К ним более не надо ничего добавлять. Так как классы pydantic.BaseModel позволяют получать json-schema, то остается только сделать скрипт сборки документации (если кратко, то надо: перед запуском приложения пройтись по всем обработчикам и у каждого заменить докстринг на swagger-описание, построенное на основе уже имеющегося докстринга и json-схем входящих данных и возврата).

И я эту документацию собираю. Но не стал рассказывать про это в статье. Причина в том, что я не нашел библиотеки для swagger и aiohttp, которая бы работала полностью как надо (или я не нашел способа заставить работать как надо).

Например, библиотека aiohttp-swagger некорректно отображает аргумент (в областях с примерами), если в аннотации есть алиас Union.

Библиотека aiohttp-swagger3, напротив, все прекрасно показывает, но не работает если в приложении есть sub_app.

Если кто-то знает как решить эти проблемы, или, возможно, кто-то знает библиотеку, которая работает стабильно — буду очень благодарен за комментарий.

В итоге у нас имеются три класса для json middleware с разными возможностями. Любой из них можно изменить под свои нужды. Или создать на их основе новый.

Можно создавать любые оболочки для содержимого запросов и ответов. Так же, можно гибко настраивать валидацию, и применять ее только там, где она действительно необходима.

Не сомневаюсь в том, что примеры которые я предложил, можно реализовать и по другому. Но надеюсь, что мои решения, если и не пригодятся полностью, то поспособствуют нахождению иных, более подходящих.

Спасибо за уделенное время. Буду рад замечаниям, и уточнениям.