Last active
April 22, 2024 04:02
-
-
Save ghandic/21c27470f6797dd856208a2c68f3e43a to your computer and use it in GitHub Desktop.
fastapi sortby/filter
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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]: | |
| ... |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Nice example ✌️!