如何在FastAPI中使用Pydantic模型处理表单数据?

51

我正在尝试从HTML表单提交数据并使用Pydantic模型进行验证。

使用以下代码:

from fastapi import FastAPI, Form
from pydantic import BaseModel
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>'''


class SimpleModel(BaseModel):
    no: int
    nm: str = ""

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

然而,我遇到了HTTP错误:“422 Unprocessable Entity”

{
    "detail": [
        {
            "loc": [
                "body",
                "form_data"
            ],
            "msg": "field required",
            "type": "value_error.missing"
        }
    ]
}

Firefox生成的等效curl命令是

curl 'http://localhost:8001/form' -H 'Content-Type: application/x-www-form-urlencoded' --data 'no=1&nm=abcd'

这里的请求主体包含no=1&nm=abcd

我做错了什么?


看起来请求体是空的,或者至少缺少 form_data。但是如果不看你提交了什么,就无法提供更多帮助。 - SColvin
在上面的代码中,GET请求提供了一个HTML表单,我点击提交。无论我输入什么值,都会出现错误。 - shanmuga
找出问题发生的第一步是检查POST请求并查看提交了什么。 - SColvin
请求正文包含 no=1&nm=abcd - shanmuga
1
请参考以下答案:这个这个 - Chris
9个回答

66
我找到了一个解决方案,可以帮助我们在FastAPI表单中使用Pydantic :)

我的代码:

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

不确定这有什么帮助。你能给一个简短的工作示例吗? - shanmuga
现在来看一下。 - Nikita Davydov
在Swagger UI中,该如何命名请求体模型?这是我想使用Pydantic类的唯一原因。 - The Fool
这个解决方案在 fastapi 0.103.0 版本中无法正常工作。 - oholimoli
快速API 0.103.1的解决方案:https://stackoverflow.com/a/77113651/7433128 - undefined

14

您可以使用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


这个解决方案看起来最简洁,但在数据验证速度方面它是最快的方法吗? - William Le
有趣,我会对比一下这个和 Pydantic Dataclass 的速度,并进行更新。 - Irfanuddin
5
数据验证速度,真的吗?你真的在意减少HTTP请求的几毫秒吗?我只熟悉极少数情况下这会有所影响。 - ron rothman
只是好奇,继承BaseModel pydantic类有什么优势吗? - undefined
1
@TedStresen-Reuter,是的,Pydantic 提供了一些优势,比如更快的序列化、对特定数据类型的支持、数据验证等等。 - undefined

8

我实现了在这里找到的解决方案Mause solution,似乎它能够工作。

from fastapi.testclient import TestClient
from fastapi import FastAPI, Depends, Form
from pydantic import BaseModel


app = FastAPI()


def form_body(cls):
    cls.__signature__ = cls.__signature__.replace(
        parameters=[
            arg.replace(default=Form(...))
            for arg in cls.__signature__.parameters.values()
        ]
    )
    return cls


@form_body
class Item(BaseModel):
    name: str
    another: str


@app.post('/test', response_model=Item)
def endpoint(item: Item = Depends(Item)):
    return item


tc = TestClient(app)


r = tc.post('/test', data={'name': 'name', 'another': 'another'})

assert r.status_code == 200
assert r.json() == {'name': 'name', 'another': 'another'}

6

您可以像下面这样使用data-form:

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

谢谢您的回答,但这并没有帮助到我。我正在寻求具体用法。我试图避免添加任何额外的代码来增加复杂性。此外,我计划将其与从表单提交的其他简单变量/文件混合使用。类似于使用“Path”或“Body”可以完成的操作。 - shanmuga

3

如果你只是想将表单数据抽象成一个类,那么你可以使用普通的类来实现

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 Model。

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, any_param, any_other_param)

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

3
我对Nikita Davydov的提案进行了一些修改,以使验证器能够在Pydantic2和FastAPI 0.103.1中正常工作。
def as_form(cls):
    new_params = [
        inspect.Parameter(
            field_name,
            inspect.Parameter.POSITIONAL_ONLY,
            default=model_field.default,
            annotation=Annotated[model_field.annotation, *model_field.metadata, Form()],
        )
        for field_name, model_field in cls.model_fields.items()
    ]

    cls.__signature__ = cls.__signature__.replace(parameters=new_params)

    return cls
使用方法:
def before_validate_int(value: int) -> int:
    raise ValueError('before int')


MyInt = Annotated[int, BeforeValidator(before_validate_int)]


@as_form
class User(BaseModel):
    age: MyInt


@app.post("/postdata")
def postdata(user: User = Depends()):
    return {"age": user.age}
尝试发布数据时,如预期一样返回了数据验证错误。
{
  "detail": [
    {
      "type": "value_error",
      "loc": [
        "body",
        "age"
      ],
      "msg": "Value error, before int",
      "input": "12",
      "ctx": {
        "error": {}
      },
      "url": "https://errors.pydantic.dev/2.3/v/value_error"
    }
  ]
}


我尝试了你提出的解决方案,但是我得到了一个错误:SyntaxError: invalid syntax. Perhaps you forgot a comma?,出现在as_form的定义中。 - undefined
我看不到哪里需要加逗号。我在Python 3.11中工作。 - undefined
pyflakes 报告了相同的 pyflakes:Error:invalid syntax. Perhaps you forgot a comma? 在这一行上: annotation=Annotated[model_field.annotation, *model_field.metadata, Form()]然而,它仍然按预期工作。 :) - undefined
是的,这是行尾的逗号。是黑色添加的。你可以删除它。 - undefined

2

按照以下方式创建类:

from fastapi import Form

class SomeForm:

    def __init__(
        self,
        username: str = Form(...),
        password: str = Form(...),
        authentication_code: str = Form(...)
    ):
        self.username = username
        self.password = password
        self.authentication_code = authentication_code


@app.post("/login", tags=['Auth & Users'])
async def auth(
        user: SomeForm = Depends()
):
    # return something / set cookie

结果:

结果

如果你想要从 JavaScript 发送一个 HTTP 请求,你必须使用 FormData 来构建请求:

const fd = new FormData()
fd.append('username', username)
fd.append('password', password)

axios.post(`/login`, fd)

你的回答可以通过提供更多支持信息来改进。请编辑以添加进一步的细节,例如引用或文档,以便他人可以确认你的答案是正确的。您可以在帮助中心中找到有关如何编写良好答案的更多信息。 - Community

2
更新了对Zac Stucke的答案,使其适用于Pydantic2(并支持mypy)。
from typing import Any
import inspect

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

api = FastAPI()

class FormBaseModel(BaseModel):
    @classmethod
    def __pydantic_init_subclass__(cls, *args: Any, **kwargs: Any) -> None:
        super().__pydantic_init_subclass__(*args, **kwargs)
        new_params = []
        schema_params = []
        for field_name, field in cls.model_fields.items():
            field_default = Form(...)
            new_params.append(
                inspect.Parameter(
                    field_name,
                    inspect.Parameter.POSITIONAL_ONLY,
                    default=Form(field.default) if not field.is_required() else field_default,
                    annotation=inspect.Parameter.empty,
                )
            )
            schema_params.append(
                inspect.Parameter(
                    field_name,
                    inspect.Parameter.POSITIONAL_ONLY,
                    default=Form(field.default) if not field.is_required() else field_default,
                    annotation=field.annotation,
                )
            )

        async def _as_form(**data: dict[str, Any]) -> BaseModel:
            try:
                return cls(**data)
            except ValidationError as e:
                raise RequestValidationError(e.raw_errors)

        async def _schema_mocked_call(**data: dict[str, Any]) -> None:
            """
            A fake version which is given the actual annotations, rather than typing.Any,
            this version is used to generate the API schema, then the routes revert back to the original afterwards.
            """
            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(  # type: ignore
            parameters=schema_params
        )
        # Set the schema patch func as an attr on the _as_form func so it can be accessed later from the route itself:
        setattr(_as_form, "_schema_mocked_call", _schema_mocked_call)

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

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

1

简而言之:一个符合 mypy 标准的可继承版本,能够生成正确的 OpenAPI 模式字段类型,而不是任何/未知类型。

现有的解决方案将 FastAPI 参数设置为 typing.Any,以防止验证两次失败,这会导致生成的 API 规范具有这些表单字段的任何/未知参数类型。

此解决方案在模式生成之前暂时注入正确的注释,并在其他解决方案中重置它们。

# Example usage
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):
            """
            A fake version which is given the actual annotations, rather than typing.Any,
            this version is used to generate the API schema, then the routes revert back to the original afterwards.
            """
            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
        # Set the schema patch func as an attr on the _as_form func so it can be accessed later from the route itself:
        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

    # The functions to call after schema generation to reset the routes to their original state:
    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 it's a form dependency, set the annotations to their true values:
                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,
                        # This mocked func was set as an attribute on the original, correct function,
                        # replace it here temporarily:
                        call=dependency.call._schema_mocked_call,  # type: ignore
                        name=dependency.name,
                        security_scopes=dependency.security_scopes,
                        use_cache=False,  # Overriding, so don't want cached actual version.
                    )

            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]

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,