python 如何在FastAPI中使用Pydantic模型和Form数据?

sqougxex  于 2023-09-29  发布在  Python
关注(0)|答案(9)|浏览(269)

我尝试从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不可处理的实体”

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

等效的curl命令(由Firefox生成)是

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

这里的请求体包含no=1&nm=abcd
我做错了什么?

ogq8wdun

ogq8wdun1#

我发现了一个解决方案,可以帮助我们使用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)):
        ...

它在《招摇》中表现为一种常见的形式。
作为装饰器,它可以更通用:

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
h22fl7wq

h22fl7wq2#

使用数据类可以更简单地完成此操作

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
n6lpvg4x

n6lpvg4x3#

我实现了这里找到的解决方案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'}
6ju8rftf

6ju8rftf4#

你可以使用下面的数据表单:

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

qojgxg4l5#

如果您只考虑将表单数据抽象到一个类中,那么可以使用普通类来实现

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

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

iqxoj9l96#

以这种方式创建类:

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)
a6b3iqyw

a6b3iqyw7#

我对proposal of Nikita Davydov做了一点修改,使验证器可以在Pydatic2和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"
    }
  ]
}
7ivaypg9

7ivaypg98#

Tldr:其他解决方案的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]
z6psavjg

z6psavjg9#

更新了Zac Stucke的答案,使其解决方案可以在Pydantic2中工作(以及mypy支持)
# form_utils.py

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]

相关问题