Превращаем любой фреймворк в FastAPI. На прицеле Flask

Я уверен, вы знаете, что такое FastAPI. По результатам опросов Stackoverflow этот фреймворк уверенно входит в топ любимых фреймворков не только питонистов, но и разработчиков в целом. И не зря: за счет свеого подхода к сериализации данных он предоставляет действительно потрясающий опыт разработки.

На Хабре уже была отличная статья о том, как устроен FastAPI изнутри. Там достаточно подробно рассмотрены все ключевые концепции этого фреймворка, а также разобраны детали его реализации на базе Starlette.

В этой же статье я хочу поделиться некоторым опытом создания своего инструмента для сериализации данных на основе pydantic. После прочтения этой статьи вы сможете превратить практически любой Python фреймворк в идейное подобие FastAPI. Причем речь идет не только об HTTP фреймворках. Никто не запрещает вам сделать, например, фреймворк для создания CLI, или telegram-ботов, или того, чего ваша душа пожелает.

Интересно? Тогда добро пожаловать!

Что мы хотим получить?

Для начала стоит определиться, что мы понимаем под подобием FastAPI. Я закладываю сюда следующие критерии:

• сериализация входящих данных на основе аннотации типов с использованием pydantic

• удобная и простая в использовании система внедрения зависимостей

Также к авторским особенностям FastAPI, возможно, стоит отнести автоматическую генерацию OpenAPI схемы (хотя и сам @tiangolo говорит, что позаимствовал эту идею у Django Rest Framework). Однако, это не та фича, которая подойдет любому фреймворку. Не все же фреймворки HTTP.

Все остальное в FastAPI — это просто набор полезных батареек и оберток над Starlette. Так что будем считать, что реализуя вышеописанные пункты мы получаем идейное продолжение FastAPI.

Надеюсь, с этим определились.

Как мы хотим получить это?

В статье, указанной в начале, уже описана эта концепция и даже есть пример реализации, взятый из исходников FastAPI.

Идея лежит на поверхности. Просто посмотрите на два следующих куска кода:

def func(     user_id: int,     username: str, ):     pass 

И вот этот:

from pydantic import BaseModel  class FuncArguments(BaseModel):     user_id: int     username: str 

Не замечаете никакого сходства? В любом случае, @tiangolo заметил. И сделал очень простую вещь: декоратор, который анализирует аннотацию функции и строит pydantic модель на ее основе.

Ну а дальше мы просто делаем немного магии в стиле

func(**FuncArguments(user_id=1, username="john").dict()) 

И вот уже наша функция вызывается с провалидированными с помощью pydantic аргументами.

Итак, псевдокод нашего декоратора выглядит следующим образом:

from functools import wraps  def validate(func):     pydantic_args_model = build_pydantic_model(func)      @wraps(func)     def validate_wrapper(*args, **kwargs):         arguments = merge_args_and_kwargs(args, kwargs)         validated_args = pydantic_args_model(**arguments).dict()         return func(**validated_args)          return validate_wrapper 

Плюсом данного подхода является то, что построение моделей происходит на этапе декорирования функции.
Во время рантайма у нас происходит только валидация pydantic модели (что быстро благодаря Rust) и вызов оригинальной функции.
Оверхед минимален.

Осталось только реализовать две функции:

• build_pydantic_model

• merge_args_and_kwargs

Детали их реализации останутся за рамками статьи, ничего сложного там нет (особенно во второй). При большом желании вы можете ознакомиться с исходниками.

В результате, мы имеем проект FastDepends, который позволяет нам превратить любой фреймворк в FastAPI.

Функционал заключается в следующем:

• валидация входящих и исходящих аргументов функции с помощью pydantic

• Depends, идентичные натуральным fastapi.Depends

• Возможность писать собственные поля — расширения

• поддержка как синхронного, так и асинхронного режимов работы

Flask + FastDepends

Попробуем разобраться на примере, что это нам дает.

Например, возьмем базовое приложение Flask и добавим туда декоратор inject из FastDepends.

from flask import Flask from fast_depends import inject, Depends from pydantic import Field, PositiveInt  app = Flask(__name__)  def get_user(user_id: PositiveInt = Field(..., alias="id")):     return f"user-{user_id}"  @app.get("/<id>") @inject def hello(user: str = Depends(get_user)):     return f"<p>Hello, {user}!</p>" 

Оно уже работает! Теперь наш handler ожидает положительный id, получает по нему пользователя внутри Depends и передает этого пользователя внутрь самой ручки.

Обработка ошибок валидации

В таком виде Flask будет возвращать 500-ки при ошибках валидации, т.к. он не знает, что такое pydantic.ValidationError. Давайте его научим?

from functools import wraps from pydantic import ValidationError  ...  def catch_validation_error(handler):     @wraps(handler)     def catch_wrapper(*args, **kwargs):         try:             return handler(*args, **kwargs)         except ValidationError as e:             # возвращаем BadRequest с подробным описанием ошибки             return e.json(), 400       return catch_wrapper  @app.get("/<id>") @catch_validation_error @inject def hello(user: str = Depends(get_user)):     return f"<p>Hello, {user}!</p>" 

Убираем матрешку

Такая куча декораторов не кажется вам немного монструозной? Я предлагаю немного их спрятать.

from flask import Flask from flask.scaffold import setupmethod  class FastFlask(Flask):     @setupmethod     def route(self, rule: str, **options):         def decorator(f):             f = catch_validation_error(inject(f))  # прячем все сюда             endpoint = options.pop("endpoint", None)             self.add_url_rule(rule, endpoint, f, **options)             return f         return decorator  app = FastFlask(__name__)  ...  @app.get("/<id>") def hello(user: str = Depends(get_user)):     return f"<p>Hello, {user}!</p>" 

Теперь намного лучше!

Обрабатываем другие виды параметров

Сейчас наш FastFlask умеет обрабатывать только параметры пути. Давайте добавим в него поддержку POST - JSON запросов, например.

Для этого в FastDepends есть специальный класс, метод use которого позволяет полностью модифицировать входящие аргументы вашей функции.
Таким образом вы можете добавлять, убирать, модифицировать набор аргументов, которые попадут в вашу функцию.

from flask import request from fast_depends.library import CustomField  class Body(CustomField):     def use(self, **kwargs):         return {             **super().use(**kwargs),             self.param_name: (request.json or {}).get(self.param_name)         }  ...  @app.post("/") def hello(user_id: PositiveInt = Body()):     # ожидает JSON вида { "user_id": 1 }     return { "Hi!": user_id } 

Итак, у нас получилось модифицировать Flask таким образом, чтобы он умел валидировать входящие аргументы с помощью pydantic, корректно обрабатывать ошибки валидации, а также имел поддержку механизма внедрения зависимостей через Depends.

Достаточно похоже на FastAPI, не находите?

Аналогичным образом вы можете поступить с любым HTTP (и не только) Python фреймворком.

Итоговый код приложения вы можете посмотреть здесь

Зачем?

В целом, основная идея проекта — дать удобный инструментарий для разработчиков OSS проектов.

Например, вы можете ознакомиться с моим фреймворком для разработки event-driven микросервисов Propan, который полностью утилизирует все особенности FastDepends чтобы максимально приблизить опыт разработки к опыту, который дает вам FastAPI.

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