fastapi sortby/filter

fastapi_query_utils.py

from enum import auto
from typing import List, Optional
from collections import Counter

from fastapi_utils.enums import StrEnum
from fastapi_utils.inferring_router import InferringRouter
from pydantic import BaseModel
from fastapi import HTTPException, Query, Depends

router = InferringRouter()


class Record(BaseModel):
    ...


class SortOption(StrEnum):
    desc = auto()
    asc = auto()
    NONE = auto()


class SortBy(BaseModel):

    name: str
    by: SortOption

    @staticmethod
    def create(options: List[str]) -> str:
        return "|".join(f"^{opt}$|^{opt}\:asc$|^{opt}\:desc$" for opt in options)

    @staticmethod
    def parse(value: str) -> "SortBy":
        vals = value.split(":", 1)
        return SortBy(name=vals[0], by=vals[1:] or SortOption.NONE)


class FilterBy(BaseModel):

    name: str
    q: str

    @staticmethod
    def create(options: List[str]) -> str:
        return "|".join(f"^{opt}\:.*$" for opt in options)

    @staticmethod
    def parse(value: str) -> "FilterBy":
        vals = value.split(":", 1)
        return FilterBy(name=vals[0], q=vals[1] or "")


def parse_sort(options: List[str]):
    async def _parse_sort(sort: Optional[str] = Query(None, regex=SortBy.create(options))):
        if sort:
            return SortBy.parse(sort)

    return _parse_sort


def parse_filter(options: List[str]):
    async def _parse_filter(filter: List[str] = Query(None, regex=FilterBy.create(options))):
        if not isinstance(filter, list):
            return []

        resp = [FilterBy.parse(f) for f in filter]

        counter = Counter([f.name for f in resp])
        if counter.most_common(1)[0][1] > 1:
            raise HTTPException(status_code=400, detail="Duplicate query parameters supplied for filter")
        return resp

    return _parse_filter


@router.get("/search")
async def search(
    sort: Optional[SortBy] = Depends(parse_sort(["table", "database"])),
    filter: List[FilterBy] = Depends(parse_filter(["table", "database", "column"])),
    limit: int = Query(10, ge=1, le=50),
    page: int = Query(1, ge=1),
) -> List[Record]:
    ...

Nice example ✌️!