Фильтрация и пагинация в FastAPI

Недавно столкнулся с задачей написать фильтрацию на FastAPI, пошёл гуглить и нашёл замечательную библиотеку fastapi-filter, которая сильно упрощает задачу. О ней в этой статье и пойдёт речь, а также заодно покажу простой способ пагинации без библиотек.

Приступим!

1. Фильтрация

Для начала установим саму библиотеку:

pip install fastapi_filter

Сами фильтры пишутся в виде моделей, и к классам‑фильтрам применимы все те же действия, что и к BaseModel из PyDantic, так как они от него наследуется. Поэтому я буду ниже использовать свойство alias для преобразования имени поля в camelCase и другие возможности BaseModel.

В моём случае такая модель:

class ProductType(str, Enum):     VEGETABLE = "VEGETABLE"     FRUIT = "FRUIT"   class Product(SQLModel, table=True):     __tablename__ = "products"     id: Optional[UUID] = Field(primary_key=True, default_factory=uuid4)     name: str     type: ProductType     production_date: datetime     quantity: int

В этой модели я хочу фильтроваться по 4 полям с различными типами данных.

Класс‑фильтр имеет следующий вид:

class ProductFilter(Filter):     name__in: Optional[list[str]] = Field(alias="names")     type__not_in: Optional[list[ProductType]] = Field(alias="types")     production_date__gte: Optional[datetime] = Field(alias="productionDatesFrom")     quantity__lte: Optional[int] = Field(alias="quantityTo")      class Constants(Filter.Constants):         model = Product      class Config:         allow_population_by_field_name = True

Параметр фильтрации поля задаётся в виде filed_name__parameter.

Внутри главного класса ProductFilter есть 2 подкласса Constants и Config.

В Constants мы передаём в model ссылку на нашу модель базы данных, а в Config передаём некие параметры модели, в моём случае это allow_population_by_field_name = True, он нужен для отмены блокировки изменения исходных имён полей alias»ами.

Полный список параметров фильтрации применяемых к полям:

• neq – искомое значение не равно переданному;

• gt – больше переданного;

• gte – больше или равно переданному;

• in – все записи, которые сходятся с переданными;

• isnull – фильтруемое поле пустое(null);

• lt -меньше переданного;

• lte — меньше или равно переданному;

• not_in/nin — все записи, которые не сходятся с переданными;

• like/ilike – поиск по подстроке.

В эндпоинте передаём этот фильтр как параметр функции, обёрнутый в FilterDepends, и внутри уже возвращаем сервис, которому передаём фильтр на вход:

from fastapi import APIRouter from sqlmodel import Session from fastapi_filter import FilterDepends  from product_schema import ProductFilter import sql_engine_service from product_service import ProductService  sql_engine = sql_engine_service.get_engine() router = APIRouter(prefix="/products", tags=["products"]) service = ProductService(Session(sql_engine))   @router.get("/") def get_product(product_filter: ProductFilter = FilterDepends(ProductFilter)) -> list:     return service.get_products_filter(product_filter) 

И в самом сервисе пишем простой запрос на фильтрацию, в качестве первого аргумента передаём SQLModel select с колонками, которые хотим вернуть в ответе, если нужны все колонки, то на месте первого аргумента select»а передаём саму модель, как в примере ниже:

class ProductService:      def __init__(self, session: Session) -> None:         self.session = session      def get_products_filter(self, product_filter: ProductFilter) -> list:         query_filter = product_filter.filter(select(Product))          return self.session.exec(query_filter).all()

И теперь всё готово к работе!

Запускаем FastAPI и смотрим в документацию Swagger:

1. Таким запросом делаем выбор всех продуктов кроме одного:

Получаем:

[   {      "name": "Banana",     "production_date": "2023-02-01T14:00:00",     "id": "8ced03ce-f536-4613-b90f-d17dc55fa087",     "type": "FRUIT",     "quantity": 12   },      {      "name": "Potato",     "production_date": "2023-01-01T14:00:00",     "id": "9fc3706c-9223-4e24-93de-a79df509952e",     "type": "VEGETABLE",     "quantity": 4   },      {      "name": "Apple",     "production_date": "2023-02-01T17:53:00",     "id": "31d46b0f-4891-4f9b-90bc-e5edebcea019",     "type": "FRUIT",     "quantity": 21   },      {      "name": "Peach",     "production_date": "2023-02-01T20:41:00",     "id": "8bffe4bf-278d-46fa-ae2c-0e44db9fa624",     "type": "FRUIT",     "quantity": 0   } ]

2. Применим все фильтры для получения результата их совместной работы:

Получаем:

[   {      "name": "Peach",     "production_date": "2023-02-01T20:41:00",     "id": "8bffe4bf-278d-46fa-ae2c-0e44db9fa624",     "type": "FRUIT",     "quantity": 0   } ]

Так как это единственная запись соответствующая параметрам:

Имя: из списка Potato,Banana,Apple,Peach

Тип: не VEGETABLE

Дата производства: от 2023–02–01T16:00:00

Количество: меньше или равно чем 15

2. Пагинация

А теперь ещё добавим пагинацию, так как очень часто требуется их совместное использование с фильтрацией.

Для начала добавим поля page и size в эндпоинт и предадим их на вход в сервис:

from fastapi import Query   @router.get("/") def get_product(product_filter: ProductFilter = FilterDepends(ProductFilter),                 page: int = Query(ge=0, default=0),                 size: int = Query(ge=1, le=100)) -> list:     return service.get_products_filter(product_filter, page, size)

Для того чтобы задать ограничение ввода значений, передаваемых в поле, я использовал Query импортированный из fastapi.

Поле page должно быть больше или равно 0, а также 0 это его значение по умолчанию.

Поле size должно быть больше или равно 1 и меньше или равно 100, значения по умолчанию нет, поэтому это поле обязательное.

Теперь идём в файл сервиса и добавляем там функционал пагинации:

Вначале высчитываем offset значения, это наши будущие индексы, которые мы будем использовать в срезах списков, полученных после фильтрации:

offset_min = page * size offset_max = (page + 1) * size

Итоговый сервис будет выглядеть так:

def get_products_filter(self, product_filter: ProductFilter,                         page: int, size: int) -> list:     offset_min = page * size     offset_max = (page + 1) * size     query_filter = product_filter.filter(select(Product))     filtered_data = self.session.exec(query_filter).all()     response = filtered_data[offset_min:offset_max] + [         {             "page": page,             "size": size,             "total": math.ceil(len(filtered_data) / size) - 1,         }     ]      return response

Здесь мы в filtered_data получаем все отфильтрованные данные, которые нам пригодятся дальше для создания ответа.

В конце формируем ответ, который будет состоять из содержимого переменной filtered_data cо срезом [offset_min:offset_max] и данных, о том на какой мы странице находимся, какой у нас размер списка возвращаемых значений и номер последней страницы.

Переходим в Swagger и смотрим на результат:

На такой запрос получаем следующий ответ:

[   {      "name": "Banana",     "production_date": "2023-02-01T14:00:00",     "id": "8ced03ce-f536-4613-b90f-d17dc55fa087",     "type": "FRUIT",     "quantity": 12   },   {      "name": "Apple",     "production_date": "2023-02-01T17:53:00",     "id": "31d46b0f-4891-4f9b-90bc-e5edebcea019",     "type": "FRUIT",     "quantity": 21   },      {      "page": 0,     "size": 2,     "total": 1   }  ]

В ответе мы получили значение total 1, а значит у нас есть ещё одна страница, меняем в запросе значение поля page с 0 на 1 и получаем последнюю страницу:

Ответ:

[   {      "name": "Peach",     "production_date": "2023-02-01T20:41:00",     "id": "8bffe4bf-278d-46fa-ae2c-0e44db9fa624",     "type": "FRUIT",     "quantity": 0   },      {      "page": 1,     "size": 2,     "total": 1   } ]

 

P.S. Статью старался писать сухо, но с акцентом на главные моменты. Если по вашему мнению чего-то не хватает или что-то наоборот лишнее, тогда жду конструктивной критики, опираясь на которую буду писать следующие статьи для вас 🙂