Filters

Overview

The Filter class is used to filter records in a database. When we filter a query using a FilterSet, each filter sequentially takes a query object and a value argument. The query object is the base query that the Filter will modify with the specified filter, while the value argument is the value to use for filtering.

Base filters

Filter

The Filter filters records in a database by field and lookup_expr.

• field - a field in a database model that the Filter will be applied to.
• lookup_expr - a function that represents a lookup expression, such as an operator from the operator or sqlalchemy.sql.operators modules.

For example:

import operator as op
from sqlalchemy.sql import operators as sa_op
from sqlalchemy_filterset import FilterSet, Filter


class ProductFilterSet(FilterSet):
    id = Filter(Product.id, lookup_expr=op.eq)
    ids = Filter(Product.id, lookup_expr=sa_op.in_op)
    name = Filter(Product.name, lookup_expr=sa_op.ilike_op)


# filter records by id = 123
filter_params = {"id": 123}
filter_set = ProductFilterSet(session, select(Product))
query = filter_set.filter(filter_params)

Filter schema pattern:

dictpydantic

value1 = 123
value2 = "some string"

filter_params = {"field1": value1, "field2": value2}

class ProductFilterSchema(BaseModel):
    field1: int | None
    field2: str | None


value1 = 123
value2 = "some string"

filter_params = ProductFilterSchema(field1=value1, field2=value2).dict(
    exclude_unset=True
)

Resulting sql expressions:

filter_params	SQL expression
{"id": 123}	select * from product where id = 123;
{"ids": [123, 345, 678]}	select * from product where id in (123, 345, 678);
{"name": "some name"}	select * from product where name like 'some name';

Here are some examples of sql conditions to be generated by various lookup expressions:

operator	SQL expression
operator.eq	field = value
operator.ne	field != value
operator.le	field <= value
operator.lt	field < value
operator.ge	field >= value
operator.gt	field > value
sqlalchemy.sql.operators.in_op	field IN value
sqlalchemy.sql.operators.is_	field IS value
sqlalchemy.sql.operators.like_op	field LIKE value
sqlalchemy.sql.operators.ilike_op	LOWER(field) LIKE value

All available operators can be found in the following modules:
operator, sqlalchemy.sql.operators, sqlalchemy-filterset.operators

RangeFilter

The RangeFilter class is used to filter records in a database by a specific field within a specified range from start to end values.

• field - a field in a database model that the RangeFilter will be applied to.
• left_lookup_expr/right_lookup_expr - lookup expressions, such as operators from the operator module (op.ge, op.gt, op.le, op.lt).
• logic_expr - logical operator, such as and_ or or_, that is used to produce a conjunction of the border expressions.

For example:

from sqlalchemy_filterset import FilterSet, RangeFilter


class ProductFilterSet(FilterSet):
    price = RangeFilter(Product.price)
    price_outer = RangeFilter(
        Product.price,
        left_lookup_expr=op.lt,
        right_lookup_expr=op.gt,
        logic_expr=sa_op.or_
    )


# filter records by the price field within a range of 1000 to 5000
filter_params = {"price": (1000, 5000)}
filter_set = ProductFilterSet(session, select(Product))
query = filter_set.filter(filter_params)

Filter schema pattern:

dictpydantic

start_value1 = 1000
end_value1 = 5000

filter_params = {"field1": (start_value1, end_value1)}

class ProductFilterSchema(BaseModel):
    price: tuple[int | None, int | None] | None


start_value1 = 1000
end_value1 = 5000

filter_params = ProductFilterSchema(field1=(start_value1, end_value1)).dict(
    exclude_unset=True
)

Resulting sql expressions:

filter_params	SQL expression
{"price": (100, 1000)}	select * from product where price >= 100 and price <= 1000;
{"price_outer": (100, 1000)}	select * from product where price < 100 or price > 1000;

SearchFilter

SearchFilter allows you to search for a given string in specified fields.

• *fields - one or several fields in a database model that the SearchFilter will be applied to.
• lookup_expr - lookup expression, default - lower(field) like '%{value}%'.
• logic_expr - logical operator, such as and_ or or_, to produce a conjunction of search expressions.

For example:

from sqlalchemy_filterset import FilterSet, SearchFilter


class ProductFilterSet(FilterSet):
    search = SearchFilter(Product.name, Product.description)


# filter records where name or description filds contain the text 'string'
filter_params = {"search": "string"}
filter_set = ProductFilterSet(session, select(Product))
query = filter_set.filter(filter_params)

Filter schema pattern:

dictpydantic

value1 = "string1"
value2 = "string2"

filter_params = {"field1": value1, "field2": value2}

class ProductFilterSchema(BaseModel):
    field1: str | None
    field2: str | None


value1 = "string1"
value2 = "string2"

filter_params = ProductFilterSchema(field1=value1, field2=value2).dict(
    exclude_unset=True
)

Resulting sql expressions:

filter_params	SQL expression
{"search": "string"}	select * from product where lower(name) like '%string%' or lower(description) like '%string%';

Warning

• Sql expression depends on the database engine. For more information, see the official documentation sqlalchemy.

Sorting

OrderingFilter

OrderingFilter allows you to specify the order in which the results of a query should be returned by adding an order by clause to the query based on specified fields.

• **fields - one or several OrderingField that define possible sorting options.

OrderingField

• field - a field in a database model that the OrderingFilter will be applied to.
• nulls - specifies whether null values should be sorted first or last.

To apply the OrderingFilter to a query, pass it a sequence of field names and the direction of the ordering (ascending or descending by prefixing with "-").

Here's an example:

from sqlalchemy_filterset import FilterSet, OrderingFilter, OrderingField


class ProductFilterSet(FilterSet):
    ordering = OrderingFilter(
        price=OrderingField(Product.price),
        name=OrderingField(Product.name, nulls=NullsPosition.last),
        id=OrderingField(Product.id)
)


# order records by name, descending price and by id
filter_params = {"ordering": ["name", "-price", "id"]}
filter_set = ProductFilterSet(session, select(Product))
query = filter_set.filter(filter_params)

Filter schema pattern:

dictpydantic

ordering_field1 = "field1"
ordering_field2 = "-field2"
ordering_field3 = "field3"

filter_params = {"ordering": [ordering_field1, ordering_field2, ordering_field3]}

class ProductFilterSchema(BaseModel):
    ordering: list[str] | None


ordering_field1 = "field1"
ordering_field2 = "-field2"
ordering_field3 = "field3"

filter_params = ProductFilterSchema(
    ordering=[ordering_field1, ordering_field2, ordering_field3]
).dict(exclude_unset=True)

Resulting sql expressions:

filter_params	SQL expression
{"ordering": ["price"]}	select * from product order by price;
{"ordering": ["-price", "id"]}	select * from product order by price desc, id;
{"ordering": ["name", "-price", "id"]}	select * from product order by name nulls last, price desc, id;

Pagination

LimitOffsetFilter

LimitOffsetFilter applies limit and offset pagination to a query. It is used to limit the number of results returned by the query and to specify a starting point (offset) in the results.

This is useful for paginating results when displaying them to the user, as it allows you to retrieve a specific page of results rather than retrieving all results at once.

To use LimitOffsetFilter, you pass it a query instance and a tuple containing the limit and offset values. Here is an example:

from sqlalchemy_filterset import FilterSet, LimitOffsetFilter


class ProductFilterSet(FilterSet):
    pagination = LimitOffsetFilter()


# Out of the 100 records received in total,
# only 10 records should be returned, starting from the 20th record.
filter_params = {"pagination": (10, 20)}
filter_set = ProductFilterSet(session, select(Product))
query = filter_set.filter(filter_params)

Filter schema pattern:

dictpydantic

limit_value = 0
offset_value = 10

filter_params = {"pagination": (limit_value, offset_value)}

class ProductFilterSchema(BaseModel):
    pagination: tuple[int, int] | None

limit_value = 0
offset_value = 10

filter_params = ProductFilterSchema(
    pagination=(limit_value, offset_value)
).dict(exclude_unset=True)

Resulting sql expressions:

filter_params	SQL expression
{"pagination": (10, 0)}	select * from product limit 10 offset 0;
{"pagination": (10, 10)}	select * from product limit 10 offset 10;

LimitOffsetFilter and Joined Tables: Getting Accurate Pagination Results

• When using this filter with joined tables, be aware that the join may modify the resulting set of records, which can affect the accuracy of pagination. This may include issues such as duplicate records or excluded records. It is important to take this into consideration and ensure that the join is properly set up to avoid these issues when using the LimitOffsetFilter.

• If we use the query:

  select users.*, roles.*
  from users
  left join user_roles on users.id = user_roles.user_id
  left join roles on user_roles.role_id = roles.id;
  

  and apply LimitOffsetFilter, we may end up with duplicate records. For example, if user with id 1 has 2 roles, we will get duplicate records of user 1 with different roles, which may affect the pagination results. We need to be attentive and use SubqueryExistsStrategy or modify sql query with group by or other methods to ensure that the join statement is set up correctly and it will not cause any issues with the pagination results.

Custom filters

InFilter/NotInFilter

InFilter and NotInFilter are subclasses of the Filter class and allow you to filter a field based on a list using the IN and NOT IN SQL operators, respectively. Under the hood it is a Filter with special in/not_in lookup expressions.

To use these filters, you can specify them in your FilterSet class like any other Filter and pass them a list of values to filter by. For example:

from sqlalchemy_filterset import FilterSet, InFilter, NotInFilter


class ProductFilterSet(FilterSet):
    ids = InFilter(Product.id)
    excluded_ids = NotInFilter(Product.id)


filter_params = {"excluded_ids": [1, 2, 3]}
filter_set = ProductFilterSet(session, select(Product))
result = filter_set.filter(filter_params)

Filter schema pattern:

dictpydantic

value1 = [1, 2, 3]
value2 = ["4", "5", "6"]

filter_params = {"field1": value1, "field2": value2}

class ProductFilterSchema(BaseModel):
    field1: list[int] | None
    field2: list[str] | None

value1 = [1, 2, 3]
value2 = ["4", "5", "6"]

filter_params = ProductFilterSchema(field1=value1, field2=value2).dict(
    exclude_unset=True
)

Resulting sql expressions:

filter_params	SQL expression
{"ids": [1, 2, 3]}	select * from product where id in (1, 2, 3);
{"excluded_ids": [1, 2, 3]}	select * from product where id not in (1, 2, 3);
{"ids": []}	select * from product where id IN (NULL) AND (1 != 1);

Warning

• Filtering by None value is not possible for sqlalchemy in_ operator.

• By passing an empty list, filtering will work according to the standard sqlalchemy rules: Empty IN Expressions.

BooleanFilter

BooleanFilter is a subclass of the Filter class that allows you to filter a field based on a boolean value using the =/!= SQL operators. Under the hood it is a Filter with special operator.eq and operator.ne lookup expressions.

Example:

from sqlalchemy_filterset import FilterSet, BooleanFilter


class ProductFilterSet(FilterSet):
    is_active = BooleanFilter(Product.is_active)


filter_params = {"is_active": True}
filter_set = ProductFilterSet(session, select(Product))
result = filter_set.filter(filter_params)

Filter schema pattern:

dictpydantic

value1 = True
value2 = False

filter_params = {"field1": value1, "field2": value2}

class ProductFilterSchema(BaseModel):
    field1: bool | None
    field2: bool | None

value1 = True
value2 = False

filter_params = ProductFilterSchema(field1=value1, field2=value2).dict(
    exclude_unset=True
)

Resulting sql expressions:

filter_params	SQL expression
{"is_active": True}	select * from product where is_active is true;
{"is_active": False}	select * from product where is_active is false;
{"is_active": None}	select * from product where is_active is null;

MethodFilter

MethodFilter allows to define a custom filtering behavior for a particular field.

Here's an example:

from sqlalchemy.sql import Select, and_, or_
from sqlalchemy_filterset import FilterSet, MethodFilter


class ProductFilterSet(FilterSet):
    available = MethodFilter(method="filter_available")

    @staticmethod
    def filter_available(query: Select, value: bool) -> Select:
        if value is True:
            return query.where(
                and_(Product.is_active.is_(True), Product.price > 100)
            )
        elif value is False:
            return query.where(
                or_(Product.is_active.is_(False), Product.price <= 100)
            )
        return query



filter_params = {"available": True}
filter_set = ProductFilterSet(session, select(Product))
result = filter_set.filter(filter_params)

Filter schema pattern is custom and depends on the value attribute expected in the MethodFilter filter method.

Resulting sql expressions:

filter_params	SQL expression
{"available": True}	select * from product where is_active is true and price > 100;
{"available": False}	select * from product where is_active is false or price <= 100;
{"available": None}	select * from product;

Custom filter

An alternative way to add custom filter behavior is to create a separate filter. To achieve this, a CustomFilter must extend the base filter class and override its filter method.

For example, the behavior of the filter method example above can be achieved as follows:

from typing import Any
from sqlalchemy.sql import Select, and_, or_
from sqlalchemy_filterset import FilterSet, BaseFilter


class AvailableFilter(BaseFilter):
    def filter(self, query: Select, value: Any, values: Any) -> Select:
        if value is True:
            return query.where(
                and_(Product.is_active.is_(True), Product.price > 100)
            )
        elif value is False:
            return query.where(
                or_(Product.is_active.is_(False), Product.price <= 100)
            )
        return query



class ProductFilterSet(FilterSet):
    available = AvailableFilter()


filter_params = {"available": True}
filter_set = ProductFilterSet(session, select(Product))
result = filter_set.filter(filter_params)

Filter strategy

Strategy is part of a Filter that controls how to connect the Filter expression with a query. The main target of it is filtering by related models in the most optimized way.

BaseStrategy

BaseStrategy is the simplest and default strategy. It simply connects the expression built by Filter to query by query.where method. It's the default value of filters.

# The same result
id = Filter(Product.id)
id = Filter(Product.id, strategy=BaseStrategy())

JoinStrategy

Join strategies are good for filtering by one-to-many or one-to-one relations. It joins table by given onclause. If a table with the same onclause has already been joined it will not be joined twice.

• model - a model or table which you want to join.
• oncaluse - an onclause expression that will be used for the join.
• is_outer - if True, generate LEFT OUTER join. Default False.
• is_full - if True, generate FULL OUTER join. Default False.

Usage

category_title = Filter(
    Category.title,
    strategy=JoinStrategy(
        Category, onclause=Product.category_id == Category.id
    ),
)

Example of result query:

select *
  from product
  join category on category.id = product.category_id
where category.title = 'test';

MultiJoinStrategy

Apply multiple JoinStrategy for filter. May used in cases of many-to-many filters.

Warning

Every case of many-to-many filtering is unique and requires query analysis. Do not use it MultiJoinStrategy blindly.

• *joins - sequence of JoinStrategy you want to apply.

Usage

tag_title = Filter(
    Tag.title,
    strategy=MultiJoinStrategy(
        JoinStrategy(TagToProduct, onclause=Product.id == TagToProduct.right_id),
        JoinStrategy(Tag, onclause=Tag.id == TagToProduct.left_id),
    )
)

Example of result query:

select *
  from product
  join tag_to_product on product.id = tag_to_product.right_id
  join tag on tag.id = tag_to_product.left_id
where tag.title = 'test';

SubqueryExistsStrategy

This strategy is good for many-to-one relations. It makes exists subquery with onclause and filter expression in where. If the query already has exists subquery with same onclause and table, it will add filter expression to the where clause.

Usage

product_title = Filter(
    Product.title,
    strategy=SubqueryExistsStrategy(
        Product, onclause=Category.id == Product.category_id
    ),
)

Example of result query:

-- select all categories for which there is a product with the name 'test'

select *
from category
where exists(select 1
             from product
             where category.id = product.category_id
               and product.title = 'test');

Efficient Data Filtering: Using the EXISTS Clause in SQL

The exists keyword is used in the above query to optimize the performance of the query by only returning the categories that have at least one product that meets the specified criteria (in this case, a product with the title of "test").

The subquery within the exists clause only needs to return one column, in this case the constant 1, to check whether there is at least one row that meets the criteria specified in the subquery's where clause. Since it only needs to check for the existence of one row, this approach is more efficient than using a traditional join and where clause to filter the data.

Additionally, using exists allows you to check for the existence of related data without actually retrieving the related data, which makes it more efficient when you don't need to retrieve the related data but just check if it exist or not.