Недавно столкнулся с задачей написать фильтрацию на 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. Статью старался писать сухо, но с акцентом на главные моменты. Если по вашему мнению чего-то не хватает или что-то наоборот лишнее, тогда жду конструктивной критики, опираясь на которую буду писать следующие статьи для вас :)