How to define custom objects as query parameters in FastAPI (using PydanticV2 annotations)?

[Kludex]

Discussed in #10101

^{Originally posted by DSamuylov August 17, 2023}

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

import bson
from typing import Annotated

from pydantic import (
    field_validator,
    PlainSerializer,
    WithJsonSchema,
    BeforeValidator,
    TypeAdapter,
)
from fastapi import FastAPI


ObjectId = Annotated[
    bson.ObjectId,
    BeforeValidator(lambda x: bson.ObjectId(x) if isinstance(x, str) else x),
    PlainSerializer(lambda x: f"{x}", return_type=str),
    WithJsonSchema({"type": "string"}, mode="validation"),
    WithJsonSchema({"type": "string"}, mode="serialization"),
]

# Casting from str to ObjectId - WORKS
ta = TypeAdapter(
    ObjectId,
    config=dict(arbitrary_types_allowed=True),
)
ta.validate_python("5f7f9f0c2a0e0b0001d5b3d0")
ta.json_schema()

# Use ObjectId as a query parameter in FastAPI - FAILS:
app = FastAPI()

@app.get("/test")
def test(id: ObjectId) -> bool:
    return True

Description

I am trying to figure out how to update my FastAPI app after from Pydantic v1 to Pydantic v2.

Before I had code that worked perfectly fine:

class ObjectId(bson.ObjectId):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, value):
        if isinstance(value, str):
            try:
                return bson.ObjectId(value)
            except bson.errors.InvalidId:
                raise ValueError("Invalid ObjectId")
        elif isinstance(value, bson.ObjectId):
            return value
        else:
            raise TypeError("ObjectId required")


app = FastAPI()

@app.get("/test")
def test(id: ObjectId) -> bool:
    return True

With the code that I provided above, I get the following error message:

fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'bson.objectid.ObjectId'> is a valid Pydantic field type. If you are using a return type annotation that is not a valid Pydantic field (e.g. Union[Response, dict, None]) you can disable generating the response model from the type annotation with the path operation decorator parameter response_model=None. Read more: https://fastapi.tiangolo.com/tutorial/response-model/

I am a bit confused, because it seems that the problem is not with a response model (which is bool in the example), but with a custom type in query parameter.

I also checked other suggested approaches, for example:

• What is the new way to declare Mongo ObjectId with PyDantic v2.0^?
• Using bson.ObjectId in Pydantic v2

There, it is suggested an alternative approach to introduce a custom annotation type, however, it still does not work for me.

I also did not find any relevant in the FastAPI documentation about query parameters and a list of Extra Data Types.

I would highly appreciate if you could help to find a possible solution.

Operating System

macOS

Operating System Details

No response

FastAPI Version

0.101.1

Pydantic Version

2.1.1

Python Version

Python 3.11.4

Additional Context

No response

[Kludex]

There's at least one issue here: the error message says there's a problem with the response field.
At least, the problem it should point out is on the request.

[Kludex]

Related: #10109

[CharlesPerrotMinotHCHB]

Facing the same issue while trying to migrate to pydantic v2, using FastAPI 0.103.1.

Works fine with pydantic v1, but fails with v2, getting the exact same error that says the issue is within the response, and removing the non-pydantic query parameter object solves the issue.

[CharlesPerrotMinotHCHB]

If that can help, I made a very small code to reproduce @Kludex

from fastapi import FastAPI

class MyType(int):
    ...

app = FastAPI()
@app.get(path="/async")
async def get_test(
    int_input: MyType,
) -> None:
    return None

The work-around is to not create the type in the query parameters, but in the function.

async def get_test(
    int_input: int,
) -> None:
    my_input = MyType(int_input)
    return None

[DSamuylov]

This is exactly the workaround that I use. It works, but then you need to do it in every API endpoint which adds quite a lot of duplicated lines in the code...

[omi-donlimit]

Having this issue as well. I would like the more concise Pydantic v2 for ObjectId to work easily with Depends

[SDAravind]

Pydantic's BeforeValidator and AfterValidator needs to raise custom error(s) for custom type(s) in case of any validation error.

BeforeValidator(lambda x: bson.ObjectId(x) if isinstance(x, str) else x),

In the provided code above, lambda expression doesn't support to raise exceptions try to implement custom fuction that handle the validation error similar to code provided below.

Additionaly, FastAPI doesn't support to raise custom error(s) (such as InvalidId from bson.errors) for custom type(s) (such as ObjectId from bson) defined using Pydantic hence ValueError in validator function check_object_id is required. Perhaps, that would be feature request to be addressed in #FastAPI.

Version

• python: 3.11
• fastapi : 0.110.0
• pydantic: 2.6.4
• pymongo: 4.6.3

Example Code

from typing import Annotated

from bson import ObjectId
# from bson.errors import InvalidId
from fastapi import FastAPI
from pydantic import (
    BaseModel,
    BeforeValidator,
    Field,
    PlainSerializer,
    WithJsonSchema,
)

app = FastAPI()


def check_object_id(value: ObjectId | str) -> ObjectId:
    if isinstance(value, (ObjectId, str)) and ObjectId.is_valid(value):
        return ObjectId(value)

    raise ValueError(
        f"{value} is not a valid ObjectId, it must be a 12-byte input or a 24-character hex string"
    )

    # raise InvalidId(
    #     f"{value} is not a valid ObjectId, it must be a 12-byte input or a 24-character hex string"
    # )


OID = Annotated[
    ObjectId | str | None,
    Field(None),
    BeforeValidator(check_object_id),
    PlainSerializer(
        func=lambda x: None if x is None else str(x),
        return_type=str | None,
    ),
    WithJsonSchema({"type": "string"}, mode="validation"),
    WithJsonSchema({"type": "string"}, mode="serialization"),
]


class Foo(BaseModel):
    model_config = {"arbitrary_types_allowed": True}
    id: OID
    bar: str = "test"


@app.post("/bar/")
async def bar(foo: Foo):
    return foo