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

Question

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?

Answer 1

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

Answer 2

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

Answer 3

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

Answer 4

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)

Answer 5

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

Answer 6

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:

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)

Answer 7

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]