Как использовать модель Pydantic с данными формы в FastAPI?

Похоже, вы нашли отличный способ использовать Pydantic с формами FastAPI! Спасибо, что поделились вашим решением.

Вот ваш код, который выглядит очень полезно:

class AnyForm(BaseModel):
    any_param: str
    any_other_param: int = 1

    @classmethod
    def as_form(
        cls,
        any_param: str = Form(...),
        any_other_param: int = Form(1)
    ) -> AnyForm:
        return cls(any_param=any_param, any_other_param=any_other_param)

@router.post('')
async def any_view(form_data: AnyForm = Depends(AnyForm.as_form)):
        ...

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

А также вы предложили более универсальный вариант в виде декоратора:

import inspect
from typing import Type

from fastapi import Form
from pydantic import BaseModel
from pydantic.fields import ModelField

def as_form(cls: Type[BaseModel]):
    new_parameters = []

    for field_name, model_field in cls.__fields__.items():
        model_field: ModelField  # type: ignore

        new_parameters.append(
             inspect.Parameter(
                 model_field.alias,
                 inspect.Parameter.POSITIONAL_ONLY,
                 default=Form(...) if model_field.required else Form(model_field.default),
                 annotation=model_field.outer_type_,
             )
         )

    async def as_form_func(**data):
        return cls(**data)

    sig = inspect.signature(as_form_func)
    sig = sig.replace(parameters=new_parameters)
    as_form_func.__signature__ = sig  # type: ignore
    setattr(cls, 'as_form', as_form_func)
    return cls

Использование этого декоратора выглядит так:

@as_form
class Test(BaseModel):
    param: str
    a: int = 1
    b: str = '2342'
    c: bool = False
    d: Optional[float] = None

@router.post('/me', response_model=Test)
async def me(request: Request, form: Test = Depends(Test.as_form)):
    return form

Это решение действительно позволяет создавать формы более удобно и лаконично! Если у вас есть дополнительные вопросы или вы хотите улучшить что-то еще, не стесняйтесь спрашивать.

Вы можете сделать это еще проще, используя dataclasses.

from dataclasses import dataclass
from fastapi import FastAPI, Form, Depends
from starlette.responses import HTMLResponse

app = FastAPI()

@app.get("/form", response_class=HTMLResponse)
def form_get():
    return '''<form method="post"> 
    <input type="text" name="no" value="1"/> 
    <input type="text" name="nm" value="abcd"/> 
    <input type="submit"/> 
    </form>'''

@dataclass
class SimpleModel:
    no: int = Form(...)
    nm: str = Form(...)

@app.post("/form")
def form_post(form_data: SimpleModel = Depends()):
    return form_data

В приведенном примере мы используем dataclass для создания модели данных SimpleModel, которая принимает данные формы, такие как no и nm. Это упрощает обработку данных формы в FastAPI, так как позволяет автоматически валидировать и парсить входящие данные с помощью зависимости Depends().

Вы можете использовать следующий код для обработки данных формы в вашем FastAPI приложении:

@app.post("/form", response_model=SimpleModel)
def form_post(no: int = Form(...), nm: str = Form(...)):
    return SimpleModel(no=no, nm=nm)

Здесь мы определяем эндпоинт POST /form, который принимает два параметра: no типа int и nm типа str. Оба параметра извлекаются из данных формы с помощью Form(...). После этого возвращается экземпляр модели SimpleModel, в который передаются значения no и nm.

Если вы хотите абстрагировать данные формы в класс, вы можете сделать это с помощью простого класса. Вот пример:

from fastapi import Form, Depends

class AnyForm:
    def __init__(self, any_param: str = Form(...), any_other_param: int = Form(1)):
        self.any_param = any_param
        self.any_other_param = any_other_param

    def __str__(self):
        return "AnyForm " + str(self.__dict__)

@app.post('/me')
async def me(form: AnyForm = Depends()):
    print(form)
    return form

Кроме того, вы можете преобразовать его в модель Pydantic:

from uuid import UUID, uuid4
from fastapi import Form, Depends
from pydantic import BaseModel

class AnyForm(BaseModel):
    id: UUID
    any_param: str
    any_other_param: int

    def __init__(self, any_param: str = Form(...), any_other_param: int = Form(1)):
        id = uuid4()
        super().__init__(id=id, any_param=any_param, any_other_param=any_other_param)

@app.post('/me')
async def me(form: AnyForm = Depends()):
    print(form)
    return form

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

Кратко: Это решение совместимо с mypy и позволяет наследовать функциональность других решений, генерируя правильные типы полей в схеме OpenAPI, а не типы Any или unknown.

Существующие решения устанавливают параметры FastAPI как typing.Any, чтобы избежать двойной валидации и последующего краха, что приводит к тому, что сгенерированная спецификация API имеет типы параметров Any или unknown для этих полей формы.

Это решение временно добавляет правильные аннотации к маршрутам перед генерацией схемы и сбрасывает их в соответствии с другими решениями после этого.

# Пример использования
class ExampleForm(FormBaseModel):
    name: str
    age: int

@api.post("/test")
async def endpoint(form: ExampleForm = Depends(ExampleForm.as_form)):
    return form.dict()

form_utils.py

import inspect
from pydantic import BaseModel, ValidationError
from fastapi import Form
from fastapi.exceptions import RequestValidationError

class FormBaseModel(BaseModel):

    def __init_subclass__(cls, *args, **kwargs):
        field_default = Form(...)
        new_params = []
        schema_params = []
        for field in cls.__fields__.values():
            new_params.append(
                inspect.Parameter(
                    field.alias,
                    inspect.Parameter.POSITIONAL_ONLY,
                    default=Form(field.default) if not field.required else field_default,
                    annotation=inspect.Parameter.empty,
                )
            )
            schema_params.append(
                inspect.Parameter(
                    field.alias,
                    inspect.Parameter.POSITIONAL_ONLY,
                    default=Form(field.default) if not field.required else field_default,
                    annotation=field.annotation,
                )
            )

        async def _as_form(**data):
            try:
                return cls(**data)
            except ValidationError as e:
                raise RequestValidationError(e.raw_errors)

        async def _schema_mocked_call(**data):
            """
            Фейковая версия, которая получает настоящие аннотации вместо typing.Any,
            эта версия используется для генерации схемы API, затем маршруты возвращаются обратно к исходным.
            """
            pass

        _as_form.__signature__ = inspect.signature(_as_form).replace(parameters=new_params)  # type: ignore
        setattr(cls, "as_form", _as_form)
        _schema_mocked_call.__signature__ = inspect.signature(_schema_mocked_call).replace(parameters=schema_params)  # type: ignore
        # Устанавливаем функцию мокирования схемы как атрибут для функции _as_form, чтобы к ней можно было обратиться позже:
        setattr(_as_form, "_schema_mocked_call", _schema_mocked_call)

    @staticmethod
    def as_form(parameters=[]) -> "FormBaseModel":
        raise NotImplementedError

# asgi.py

from fastapi.routing import APIRoute
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from fastapi.dependencies.utils import get_dependant, get_body_field

api = FastAPI()

def custom_openapi():
    if api.openapi_schema:
        return api.openapi_schema

    def create_reset_callback(route, deps, body_field):
        def reset_callback():
            route.dependant.dependencies = deps
            route.body_field = body_field

        return reset_callback

    # Функции, которые будут вызываться после генерации схемы, чтобы сбросить маршруты в исходное состояние:
    reset_callbacks = []

    for route in api.routes:
        if isinstance(route, APIRoute):
            orig_dependencies = list(route.dependant.dependencies)
            orig_body_field = route.body_field

            is_modified = False
            for dep_index, dependency in enumerate(route.dependant.dependencies):
                # Если это зависимость формы, установить аннотации на их истинные значения:
                if dependency.call.__name__ == "_as_form":  # type: ignore
                    is_modified = True
                    route.dependant.dependencies[dep_index] = get_dependant(
                        path=dependency.path if dependency.path else route.path,
                        # Эта мокированная функция была установлена как атрибут на оригинальной, правильной функции,
                        # заменяем ее здесь временно:
                        call=dependency.call._schema_mocked_call,  # type: ignore
                        name=dependency.name,
                        security_scopes=dependency.security_scopes,
                        use_cache=False,  # Переопределяем, так что не хотим кэшированную актуальную версию.
                    )

            if is_modified:
                route.body_field = get_body_field(dependant=route.dependant, name=route.unique_id)

                reset_callbacks.append(
                    create_reset_callback(route, orig_dependencies, orig_body_field)
                )

    openapi_schema = get_openapi(
        title="foo",
        version="bar",
        routes=api.routes,
    )

    for callback in reset_callbacks:
        callback()

    api.openapi_schema = openapi_schema
    return api.openapi_schema

api.openapi = custom_openapi  # type: ignore[assignment]

Если у вас остались вопросы или вам нужно уточнить что-то еще, не стесняйтесь спрашивать!