Allow custom type for Query parameters

[StephenPCG]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue 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

from typing import Optional
from fastapi import APIRouter, Query
from datetime import datetime

router = APIRouter()

class DateTimeRange:
    start: datetime
    end: datetime

    def __init__(self, value: str):
        # value supposed to be in format: `2021-01-01,2021-02-01`
        start, end = value.split(",")
        self.start = datetime.strptime(start, "%Y-%m-%d")
        self.end = datetime.strptime(end, "%Y-%m-%d")

@router.get("/messages")
def api_filter_message(
    created_at_range: Optional[DateTimeRange] = Query(None),
    modified_at_range: Optional[DateTimeRange] = Query(None),
):
    """
    GET /messages?created_at_range=...&modified_at_range=...

    List messages filtered by Message.created_at and Message.modified_at
    """
    ...

Description

The above code won't run, which complains:

fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'DateTimeRange'> is a valid pydantic field type

If I change implementation of DateTimeRange and let it be a pydantic model, FastAPI complains again, says:

Param: created_at_range can only be a request body, using Body(...)

Currently what only I can do is like following:

@router.get("/messages")
def api_filter_message(
    created_at_range_param: Optional[str] = Query(None, alias="created_at_range"),
    modified_at_range_param: Optional[str] = Query(None, alias="modified_at_range"),
):
    created_at_range = DateTimeRange(created_at_range_param) if created_at_range_param else None
    modified_at_range = DateTimeRange(updated_at_range_param) if updated_at_range_param else None
    ...

which is very annoying and repeating work.

Wanted Solution

It would be nice to have a custom validator (data converter) for Query (and possibly Path, Header, Cookie, etc, which input is always str). And then support custom types.

Wanted Code

@router.get("/messages")
def api_filter_messages(
    created_at_range: Optional[DateTimeRange] = Query(None, validator=lambda v: DateTimeRange(v)),
    modified_at_range: Optional[DateTimeRange] = Query(None, validator=lambda v: DateTimeRange(v)),
):
    ...

The argument `validator` takes idea from Pydantic's validator, which takes the input value, do validate, and convert to target type.

Alternatives

An alternative approach is to make Query accept types which implements the following protocol:

class ConvertableFromStr(Protocol):
    @classmethod
    def from_str(cls, v):
        ...

When Query sees such types, use its from_str() method to create the type instance.

Operating System

macOS

Operating System Details

No response

FastAPI Version

0.67.0

Python Version

Python 3.9.0

Additional Context

No response

[Dustyposa]

For custom validator , how about this way?
https://pydantic-docs.helpmanual.io/usage/validators/

[StephenPCG]

@Dustyposa I was not asking how to implement custom validators, but asking for if FastAPI's Query(), Cookie(), Header() and Path() can support non-scalar types. Currently only Body() support non-scalar types.

Correct me if I did'n get your point.

[Dustyposa]

Oh, I'm so sorry. But i can't get your point until now.
You pasted code have no Body() code. Can you past the code that Body() can do, other can't do in your situation?

[Mause]

You can use a custom validator to parse values, see the documentation linked above

[StephenPCG]

@Dustyposa Here is an example code:

# filename: app.py
from datetime import datetime
from functools import cached_property

from fastapi import Body, FastAPI, Query
from pydantic import BaseModel

app = FastAPI()


class DateRange(BaseModel):
    __root__: str

    class Config:
        keep_untouched = (cached_property,)

    @cached_property
    def start(self):
        start, _ = self.__root__.split(",")
        return datetime.strptime(start, "%Y-%m-%d")

    @cached_property
    def end(self):
        _, end = self.__root__.split(",")
        return datetime.strptime(end, "%Y-%m-%d")


@app.post("/")
def echo_date_range(created_at_range: DateRange = Body(..., embed=True)):
    return dict(start=created_at_range.start, end=created_at_range.end)

run with uvicorn app:app, then send an request:

$ curl -X 'POST' \
  'http://127.0.0.1:8000/' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "created_at_range": "2021-01-01,2021-02-01"
}'
> {"start":"2021-01-01T00:00:00","end":"2021-02-01T00:00:00"}

When use Body(), it can take a str value and convert to a pydantic model handy for later use. If replace Body() with Query(), it failed with the following exception:

AssertionError: Param: created_at_range can only be a request body, using Body(...)

[StephenPCG]

The following code shows another way to implement the custom data type using __get_validators__() protocol recognized by pydantic, which is also supported by Body(), but not by Query(). This code also contains more detailed explanations on why I need this:

from datetime import datetime
from typing import Optional

from fastapi import Body, FastAPI, Query
from dataclasses import dataclass

app = FastAPI()


@dataclass
class AlignedDateRange:
    start: Optional[datetime]
    end: Optional[datetime]

    @classmethod
    def validate(cls, v):
        aligned_start: Optional[datetime] = None
        aligned_end: Optional[datetime] = None

        start, end = v.split(",")
        if start:
            # align start to beginning of day
            aligned_start = datetime.strptime(start, "%Y-%m-%d").replace(hour=0, minute=0, second=0)
        if end:
            # align end to end of day
            aligned_end = datetime.strptime(end, "%Y-%m-%d").replace(hour=23, minute=59, second=59)

        return cls(start=aligned_start, end=aligned_end)

    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def __modify_schema__(cls, field_schema):
        field_schema.update(
            pattern='YYYY-MM-DD,YYYY-MM-DD',
            examples=['2021-01-01,2021-02-01'],
            type="string",
        )


@app.post("/")
def echo_date_range(created_at_range: AlignedDateRange = Body(..., embed=True)):
    """
    Use case for aligned date range:

        query = session.query(MessageModel)
        if created_at_range.start:
            query = query.filter(MessageModel.created_at >= created_at_range.start)
        if created_at_range.end:
            query = query.filter(MessageModel.created_at <= created_at_range.end)

        # filter by more timestamp fields ...

    Having an AlignedDateRange class, we can assemble the above code into a helper function:

        def filter_by_range(query: Query, model: Model, field_name: str, range: AlignedDateRange) -> Query:
            if range.start:
                query = query.filter(getattr(model, field_name) >= range.start)
            if range.end:
                query = query.filter(getattr(model, field_name) <= range.end)
            return query

    Assume this api endpoint is for listing MessageModel instances,
    with support to filter by multiple timestamp fields, the request url looks like:

        GET /messages?created_at_range=2021-01-01,2021-02-01&read_at_range=2021-01-01,2021-02-01&...

    The code for router looks like:

        @router.get("/messages", reaponse_model=List[MessageResponse])
        def list_messages(
            created_at_range: AlignedDateRange = Query(...),
            read_at_range: AlignedDateRange = Query(...),
        )
            query = session.query(Message)
            query = filter_by_range(query, MessageModel, "created_at", created_at_range)
            query = filter_by_range(query, MessageModel, "read_at", read_at_range)
            return query.all()
    """
    return dict(start=created_at_range.start, end=created_at_range.end)

[Dustyposa]

I guess I know what do you want.
But in my opinion, this have double function, first is validate the data, second is convert the data.
I think this is a good way to do.
You can create a model:

@dataclass
class XXX:
    start: Optional[datetime]
    end: Optional[datetime]

use, Depends to make request data convert to be XXX data.

@app.post("/")
def echo_date_range(created_at_range: XXX = Depends(convert_func)):

[StephenPCG]

@Dustyposa I had considered your approach, but there is one problem I don't know how to solve:

@app.post("/")
def echo_date_range(created_at_range: XXX = Depends(convert_func)):

How to tell convert_func where to get input data, e.g. from Query, Cookie or Header? from which query param?

[Dustyposa]

You can get some example like here

[StephenPCG]

I have read all docs on FastAPI Dependency, but just can not think of an elegant way.

My goal is that there is just one-shot in view function to get the desired data. For example, the view code could look like this:

@router.get("/")
def view(
    created_at_range: AlignedDateRange = Depends(AlignedDateRange.from_query("created_at_range")),
    read_at_range: AlignedDateRange = Depends(AlignedDateRange.from_query("read_at_range")),
):
    ...

But I can't figure out how to implement such AlignedDateRange.from_query, the following is what I have tried

class AlignedDateRange:
    @classmethod
    def from_query(cls, query_field_name: str):
        # here, should return a function which signature looks like:
       def get_date_range_from_query(${query_field_name}: Query()):  # <-- how to create such a function?
           return AlignedDateRange(${query_field_name})
       return get_date_range_from_query

# other ways, like using class.__init__(), class.__call__() both have the same problem.
class AlignedDateRange:
    def __init__(self, query_field_name: str):
        self.query_field_name = query_field_name

    def __call__(self, ${query_field_name}: Query()):  # <- since FastAPI get query field name by signature name, this seems can only be hardcoded.
        ...

Currently, my local working code requires two-shot, like this:

@router.get("/")
def view(
    created_at_range_raw: str = Query(...),
    read_at_range_raw: str = Query(...),
):
    created_at_range = AlignedDateRange(created_at_range_raw)
    read_at_range = AlignedDateRange(read_at_range_raw)
    ...

or in even verbose way, for each possible query field name, create a separate depends func:

def get_created_at_range(created_at: str = Query(...)):
    return AlignedDateRange(created_at)

def get_read_at_range(read_at: str = Query(...)):
    return AlignedDateRange(created_at)

def get_xxx_range(xxx: str = Query(...)):   # <- repeat for all possible query names
    return AlignedDateRange(xxx)

@router.get("/")
def view(
    created_at_range: AlignedDateRange = Depends(get_created_at_range),
    read_at_range: AlignedDateRange = Depends(get_read_at_range),
    xxx_range: AlignedDateRange = Depends(get_xxx_range),
):
    ...

I believe this is not an elegant approach. That is why I created this issue. I believe the most elegant use case is:

@router.get("/")
def view(
    created_at_range: AlignedDateRange = Query(...),
    read_at_range: AlignedDateRange = Query(...),
    xxx_range: AlignedDateRange = Query(...),
):
    ...

[Mause]

Is this not elegant?

from fastapi import Depends, FastAPI, Query
from fastapi.testclient import TestClient
from datetime import date
from pydantic import BaseModel
from typing import Optional, TypeVar, Type, Protocol


class AlignedDateRange(BaseModel):
    start: Optional[date]
    end: Optional[date]

    @classmethod
    def validate(cls, v):
        start, end = v.split(",")

        return cls(start=start or None, end=end or None)

    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def __modify_schema__(cls, field_schema):
        field_schema.update(
            pattern="YYYY-MM-DD,YYYY-MM-DD",
            examples=["2021-01-01,2021-02-01"],
            type="string",
        )

    @classmethod
    def query(cls, name: str):
        def factory(field=Query(..., alias=name)):
            return cls.validate(field)

        return factory


app = FastAPI()


@app.get("/")
def index(ranger=Depends(AlignedDateRange.query("hello"))):
    return ranger


tc = TestClient(app)
print(tc.get("/").text)
print(tc.get("/", params={"hello": "2021-01-01,2021-02-01"}).text)
print(tc.get("/", params={"hello": ",2021-02-01"}).text)
print(tc.get("/", params={"hello": "2021-01-01,"}).text)

$ python .\3656.py
{"detail":[{"loc":["query","hello"],"msg":"field required","type":"value_error.missing"}]}
{"start":"2021-01-01","end":"2021-02-01"}
{"start":null,"end":"2021-02-01"}
{"start":"2021-01-01","end":null}

[StephenPCG]

@Mause Wow, the factory function is great! That's what I didn't come up with. Thanks a lot!