shanmuga
shanmuga

Reputation: 4499

How to use a Pydantic model with Form data in FastAPI?

I am trying to submit data from HTML forms and validate it with a Pydantic model.

Using this code

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

However, I get the HTTP error: "422 Unprocessable Entity"

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

The equivalent curl command (generated by Firefox) is

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

Here the request body contains no=1&nm=abcd.

What am I doing wrong?

Upvotes: 56

Views: 93195

Answers (10)

MatsLindh
MatsLindh

Reputation: 52802

FastAPI 0.113.0 added support for declaring Form requests as Pydantic models.

The example from the release:

from typing import Annotated

from fastapi import FastAPI, Form
from pydantic import BaseModel

app = FastAPI()


class FormData(BaseModel):
    username: str
    password: str


@app.post("/login/")
async def login(data: Annotated[FormData, Form()]):
    return data

This is further documented under "Form fields" in the reference guide. So for anyone coming across this question after September 2024, there is now a native solution built into FastAPI itself.

Upvotes: 2

Roger Collins
Roger Collins

Reputation: 1259

Updated the answer from Zac Stucke for his solution to work in Pydantic2 (and mypy support)

# 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]

Upvotes: 4

julia uss
julia uss

Reputation: 161

I changed a bit proposal of Nikita Davydov for validators to work in Pydatic2 and 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

The usage:

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}

When trying to post data validation error is returned as expected:

{
  "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"
    }
  ]
}

Upvotes: 8

Zak Stucke
Zak Stucke

Reputation: 452

Tldr: a mypy compliant, inheritable version of other solutions that produces the correct generated OpenAPI schema field types rather than any/unknown types.

Existing solutions set the FastAPI params to typing.Any to prevent the validation from occurring twice and failing, this causes the generated API spec to have any/unknown param types for these form fields.

This solution temporarily injects the correct annotations to the routes before schema generation, and resets them in line with other solutions afterwards.

# 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]

Upvotes: 1

Nikita Davydov
Nikita Davydov

Reputation: 957

I found a solution that can help us to use Pydantic with FastAPI forms :)

My code:

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)):
        ...

It's shown in the Swagger as a usual form.

It can be more generic as a decorator:

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

And the usage looks like

@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

Upvotes: 73

Irfanuddin
Irfanuddin

Reputation: 2605

You can do this even simpler using 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

Upvotes: 19

artemonsh
artemonsh

Reputation: 113

Create the class this way:

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

Result:

Result

If you want then to make an http request from javascript you must use FormData to construct the request:

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

axios.post(`/login`, fd)

Upvotes: 2

stefanlsd
stefanlsd

Reputation: 101

I implemented the solution found here Mause solution and it seemed to work

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'}

Upvotes: 10

Kassym Dorsel
Kassym Dorsel

Reputation: 4843

If you're only looking at abstracting the form data into a class you can do it with a plain class

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

And it can also be turned into a 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

Upvotes: 3

Includeamin
Includeamin

Reputation: 69

you can use data-form like below:

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

Upvotes: 4

Related Questions