Превращаем любой фреймворк в 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.

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