SQLAlchemy filters

Filter, sort and paginate SQLAlchemy query objects. Ideal for exposing these actions over a REST API.

This is fork of https://github.com/juliotrigo/sqlalchemy-filters and has only limited capabilities:

• only sqlalchemy >= 2.0 supported
• removed all restricted loads capabilities
• added filters inet_in, inet_not_in (INET columns filtering with CIDR)
• added operators astext_ilike, astext_not_ilike (naive filtering on JSON columns)

Filtering

Assuming that we have a SQLAlchemyquery object:

fromsqlalchemyimportColumn, Integer, Stringfromsqlalchemy.ext.declarativeimportdeclarative_baseclassBase(object):
id=Column(Integer, primary_key=True)
name=Column(String(50), nullable=False)
count=Column(Integer, nullable=True)
@hybrid_propertydefcount_square(self):
returnself.count*self.count@hybrid_methoddefthree_times_count(self):
returnself.count*3Base=declarative_base(cls=Base)
classFoo(Base):
__tablename__='foo'# ...query=session.query(Foo)

Then we can apply filters to that query object (multiple times):

fromsqlalchemy_filtersimportapply_filters# `query` should be a SQLAlchemy query objectfilter_spec= [{'field': 'name', 'op': '==', 'value': 'name_1'}]
filtered_query=apply_filters(query, filter_spec)
more_filters= [{'field': 'foo_id', 'op': 'is_not_null'}]
filtered_query=apply_filters(filtered_query, more_filters)
result=filtered_query.all()

It is also possible to filter queries that contain multiple models, including joins:

classBar(Base):
__tablename__='bar'foo_id=Column(Integer, ForeignKey('foo.id'))

query=session.query(Foo).join(Bar)
filter_spec= [
{'model': 'Foo', 'field': 'name', 'op': '==', 'value': 'name_1'},
{'model': 'Bar', 'field': 'count', 'op': '>=', 'value': 5},
]
filtered_query=apply_filters(query, filter_spec)
result=filtered_query.all()

apply_filters will attempt to automatically join models to query if they're not already present and a model-specific filter is supplied. For example, the value of filtered_query in the following two code blocks is identical:

query=session.query(Foo).join(Bar) # join pre-applied to queryfilter_spec= [
{'model': 'Foo', 'field': 'name', 'op': '==', 'value': 'name_1'},
{'model': 'Bar', 'field': 'count', 'op': '>=', 'value': 5},
]
filtered_query=apply_filters(query, filter_spec)

query=session.query(Foo) # join to Bar will be automatically appliedfilter_spec= [
{'field': 'name', 'op': '==', 'value': 'name_1'},
{'model': 'Bar', 'field': 'count', 'op': '>=', 'value': 5},
]
filtered_query=apply_filters(query, filter_spec)

The automatic join is only possible if SQLAlchemy can implictly determine the condition for the join, for example because of a foreign key relationship.

Automatic joins allow flexibility for clients to filter and sort by related objects without specifying all possible joins on the server beforehand. Feature can be explicitly disabled by passing do_auto_join=False argument to the apply_filters call.

Note that first filter of the second block does not specify a model. It is implictly applied to the Foo model because that is the only model in the original query passed to apply_filters.

It is also possible to apply filters to queries defined by fields, functions or select_from clause:

query_alt_1=session.query(Foo.id, Foo.name)
query_alt_2=session.query(func.count(Foo.id))
query_alt_3=session.query().select_from(Foo).add_column(Foo.id)

Hybrid attributes

You can filter by a hybrid attribute: a hybrid property or a hybrid method.

query=session.query(Foo)
filter_spec= [{'field': 'count_square', 'op': '>=', 'value': 25}]
filter_spec= [{'field': 'three_times_count', 'op': '>=', 'value': 15}]
filtered_query=apply_filters(query, filter_spec)
result=filtered_query.all()

Sort

fromsqlalchemy_filtersimportapply_sort# `query` should be a SQLAlchemy query objectsort_spec= [
{'model': 'Foo', 'field': 'name', 'direction': 'asc'},
{'model': 'Bar', 'field': 'id', 'direction': 'desc'},
]
sorted_query=apply_sort(query, sort_spec)
result=sorted_query.all()

apply_sort will attempt to automatically join models to query if they're not already present and a model-specific sort is supplied. The behaviour is the same as in apply_filters.

This allows flexibility for clients to sort by fields on related objects without specifying all possible joins on the server beforehand.

Hybrid attributes

You can sort by a hybrid attribute: a hybrid property or a hybrid method.

Pagination

fromsqlalchemy_filtersimportapply_pagination# `query` should be a SQLAlchemy query objectquery, pagination=apply_pagination(query, page_number=1, page_size=10)
page_size, page_number, num_pages, total_results=paginationassert10==len(query)
assert10==page_size==pagination.page_sizeassert1==page_number==pagination.page_numberassert3==num_pages==pagination.num_pagesassert22==total_results==pagination.total_results

Filters format

Filters must be provided in a list and will be applied sequentially. Each filter will be a dictionary element in that list, using the following format:

filter_spec= [
{'model': 'model_name', 'field': 'field_name', 'op': '==', 'value': 'field_value'},
{'model': 'model_name', 'field': 'field_2_name', 'op': '!=', 'value': 'field_2_value'},
# ...
]

The model key is optional if the original query being filtered only applies to one model.

If there is only one filter, the containing list may be omitted:

filter_spec= {'field': 'field_name', 'op': '==', 'value': 'field_value'}

Where field is the name of the field that will be filtered using the operator provided in op (optional, defaults to ==) and the provided value (optional, depending on the operator).

This is the list of operators that can be used:

• is_null
• is_not_null
• ==, eq
• !=, ne
• >, gt
• <, lt
• >=, ge
• <=, le
• like
• ilike
• not_ilike
• in
• not_in
• any
• not_any

any / not_any

PostgreSQL specific operators allow to filter queries on columns of type ARRAY. Use any to filter if a value is present in an array and not_any if it's not.

Boolean Functions

and, or, and not functions can be used and nested within the filter specification:

filter_spec= [
{
'or': [
{
'and': [
{'field': 'field_name', 'op': '==', 'value': 'field_value'},
{'field': 'field_2_name', 'op': '!=', 'value': 'field_2_value'},
]
},
{
'not': [
{'field': 'field_3_name', 'op': '==', 'value': 'field_3_value'}
]
},
],
}
]

Note: or and and must reference a list of at least one element. not must reference a list of exactly one element.

Sort format

Sort elements must be provided as dictionaries in a list and will be applied sequentially:

sort_spec= [
{'model': 'Foo', 'field': 'name', 'direction': 'asc'},
{'model': 'Bar', 'field': 'id', 'direction': 'desc'},
# ...
]

Where field is the name of the field that will be sorted using the provided direction.

The model key is optional if the original query being sorted only applies to one model.

nullsfirst / nullslast

sort_spec= [
{'model': 'Baz', 'field': 'count', 'direction': 'asc', 'nullsfirst': True},
{'model': 'Qux', 'field': 'city', 'direction': 'desc', 'nullslast': True},
# ...
]

nullsfirst is an optional attribute that will place NULL values first if set to True, according to the SQLAlchemy documentation.

nullslast is an optional attribute that will place NULL values last if set to True, according to the SQLAlchemy documentation.

If none of them are provided, then NULL values will be sorted according to the RDBMS being used. SQL defines that NULL values should be placed together when sorting, but it does not specify whether they should be placed first or last.

Even though both nullsfirst and nullslast are part of SQLAlchemy, they will raise an unexpected exception if the RDBMS that is being used does not support them.

At the moment they are supported by PostgreSQL, but they are not supported by SQLite and MySQL.

Running tests

The default configuration uses SQLite, MySQL (if the driver is installed, which is the case when tox is used) and PostgreSQL (if the driver is installed, which is the case when tox is used) to run the tests, with the following URIs:

sqlite+pysqlite:///test_sqlalchemy_filters.db
mysql+mysqlconnector://root:@localhost:3306/test_sqlalchemy_filters
postgresql+psycopg2://postgres:@localhost:5432/test_sqlalchemy_filters?client_encoding=utf8'

A test database will be created, used during the tests and destroyed afterwards for each RDBMS configured.

There are Makefile targets to run docker containers locally for both MySQL and PostgreSQL, using the default ports and configuration:

$ make mysql-container
$ make postgres-container

To run the tests locally:

$ # Create/activate a virtual environment
$ pip install tox
$ tox

There are some other Makefile targets that can be used to run the tests:

There are other Makefile targets to run the tests, but extra dependencies will have to be installed:

$ pip install -U --editable ".[dev,mysql,postgresql]"
$ # using default settings
$ make test
$ make coverage
$ # overriding DB parameters
$ ARGS='--mysql-test-db-uri mysql+mysqlconnector://root:@192.168.99.100:3340/test_sqlalchemy_filters' make test
$ ARGS='--sqlite-test-db-uri sqlite+pysqlite:///test_sqlalchemy_filters.db' make test
$ ARGS='--mysql-test-db-uri mysql+mysqlconnector://root:@192.168.99.100:3340/test_sqlalchemy_filters' make coverage
$ ARGS='--sqlite-test-db-uri sqlite+pysqlite:///test_sqlalchemy_filters.db' make coverage

Database management systems

The following RDBMS are supported (tested):

• SQLite
• MySQL
• PostgreSQL

SQLAlchemy support

The following SQLAlchemy versions are supported: 2.0.

Changelog

Consult the CHANGELOG document for fixes and enhancements of each version.

License

Apache 2.0. See LICENSE for details.