Обновленческое

Версия 0.1.5. Добавил класс PydanticHandler для поддержки POST-запросов в формате датаклассов pydantic.

Подобности тут: https://github.com/vlakir/cleanapi

Добавил в свой текущий проект. Объем кода api-сервера сократился раза в 3.

Было:

from cleanapi.server import BaseHandler
from app_logger import get_logger
from utils import process_error
import json
import pydantic
from asgiref.sync import sync_to_async
from tornado.web import RequestHandler
from abc import ABC
from common_dicts import *
from common_dataclasses import FramesRequest, FramesResult
from module_db.db_core import WorkDatabase
from utils import get_config_setting
from common_dicts import is_token_valid


url_tail = '/api/v1/frames.json'

logger = get_logger(__name__)

print_benchmark = get_config_setting('COMMON', 'print_benchmark').lower() == 'true'


class Handler(BaseHandler, RequestHandler, ABC):
    """
    Хендлер API-запроса на список систем координат
    """
    async def post(self):
        logger.info(f'Поступил входящий запрос на список систем координат')
        errors = []

        try:
            body_json_dict = json.loads(self.request.body)
        except (json.decoder.JSONDecodeError, TypeError, ValueError) as ex:
            process_error('000003', errors, str(ex), logger)
            status_code = 400
            self.set_status(status_code)
            self.write({'errors': errors})
            return

        try:
            input_request = FramesRequest(**body_json_dict)
            if not is_token_valid(input_request.token):
                raise ValueError("Некорректный токен доступа")

        except pydantic.error_wrappers.ValidationError as ex:
            process_error('000003', errors, json.loads(ex.json()), logger)
            status_code = 400
            self.set_status(status_code)
            self.write({'errors': errors})
            return

        except ValueError as ex:
            process_error('000007', errors, str(ex), logger)
            status_code = 400
            self.set_status(status_code)
            self.write({'errors': errors})
            return

        try:
            result = await sync_to_async(_process)(input_request)
        except pydantic.error_wrappers.ValidationError as ex:
            process_error('000009', errors, json.loads(ex.json()), logger)
            status_code = 400
            self.set_status(status_code)
            self.write({'errors': errors})
            return
        except Exception as ex:
            process_error('000009', errors, f'{str(ex.__class__.__name__)}: {str(ex)}', logger)
            status_code = 400
            self.set_status(status_code)
            self.write({'errors': errors})
            return
        logger.info(f'Запрос на список систем координат выполнен.')

        if len(result.errors) == 0:
            result.errors = None
            status_code = 200
        else:
            status_code = 400

        output_json = result.json(exclude_none=True)

        self.set_status(status_code)
        self.write(output_json)


# noinspection PyUnusedLocal
def _process(request: FramesRequest) -> FramesResult:
    """
    Обработчик запроса
    :param request: входящий запрос
    :type request: FramesRequest
    :return: результат обработки
    :rtype: FramesResult
    """
    errors = []
    work_database = WorkDatabase(errors, print_benchmark=print_benchmark)

    frames = work_database.get_all_frames(errors, print_benchmark=print_benchmark,
                                          benchmark_name='Получение справочника систем координат')

    result = FramesResult(frames=frames, errors=errors)

    return result

Стало:

from common_dataclasses import FramesRequest, FramesResult
from module_db.db_core import WorkDatabase
from module_frontend.app_handler import AppHandler


url_tail = '/api/v1/frames.json'


# noinspection PyAbstractClass
class Handler(AppHandler):
    """
    Хендлер API-запроса на список систем координат
    """
    request_dataclass = FramesRequest
    result_dataclass = FramesResult

    # noinspection PyUnusedLocal
    def process(self, request: request_dataclass) -> result_dataclass:
        """
        Обработчик запроса
        :param request: входящий запрос
        :type request: FramesRequest
        :return: результат обработки
        :rtype: FramesResult
        """
        self.logger.info(f'Обработка запроса на {url_tail}')

        errors = []
        work_database = WorkDatabase(errors, print_benchmark=self.print_benchmark)

        frames = work_database.get_all_frames(errors, print_benchmark=self.print_benchmark,
                                              benchmark_name='Получение справочника систем координат')

        result = FramesResult(frames=frames, errors=errors)

        return result

И так в каждом хэндлере, а их у меня там порядка двух десятков