sqlalchemy-filters 0.13.0

过滤

假设我们有一个 SQLAlchemy 查询对象

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base


class Base(object):
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    count = Column(Integer, nullable=True)

    @hybrid_property
    def count_square(self):
        return self.count * self.count

    @hybrid_method
    def three_times_count(self):
        return self.count * 3


Base = declarative_base(cls=Base)


class Foo(Base):

    __tablename__ = 'foo'

# ...

query = session.query(Foo)

然后我们可以对该查询对象应用过滤器（多次）

from sqlalchemy_filters import apply_filters


# `query` should be a SQLAlchemy query object

filter_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()

还可以过滤包含多个模型（包括连接）的查询

class Bar(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 将尝试自动将模型与 query 联接，如果它们尚未存在并且提供了模型特定的过滤器。例如，以下两个代码块中的 filtered_query 的值是相同的

query = session.query(Foo).join(Bar)  # join pre-applied to query

filter_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 applied

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

只有在 SQLAlchemy 可以隐式确定联接条件的情况下，自动联接才可行，例如，由于外键关系。

自动联接允许客户端在事先不指定所有可能的联接的情况下，对相关对象进行过滤和排序。可以通过将 do_auto_join=False 参数传递给 apply_filters 调用来显式禁用此功能。

请注意，第二个块中的第一个过滤器没有指定模型。它隐式应用于 Foo 模型，因为这是传递给 apply_filters 的原始查询中的唯一模型。

也可以将过滤器应用于由字段、函数或 select_from 子句定义的查询

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)

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()

限制加载

您可以通过使用 apply_loads 函数来限制 SQLAlchemy 从数据库中加载的字段。

query = session.query(Foo, Bar).join(Bar)
load_spec = [
    {'model': 'Foo', 'fields': ['name']},
    {'model': 'Bar', 'fields': ['count']}
]
query = apply_loads(query, load_spec)  # will load only Foo.name and Bar.count

apply_loads 函数的效果是在访问时/如果它们被访问时延迟加载任何其他字段，而不是在查询执行时加载它们。它仅适用于在正常查询执行期间会加载的字段。

query = session.query(Foo).join(Bar)
load_spec = [
    {'model': 'Foo', 'fields': ['name']}
    {'model': 'Bar', 'fields': ['count']}  # ignored
]
query = apply_loads(query, load_spec)  # will load only Foo.name

query = session.query(Foo).options(joinedload(Foo.bar))
load_spec = [
    {'model': 'Foo', 'fields': ['name']}
    {'model': 'Bar', 'fields': ['count']}
]
query = apply_loads(query, load_spec)

query = session.query(Foo).options(joinedload(Bar).load_only('count'))
load_spec = [
    {'model': 'Foo', 'fields': ['name']}
]
query = apply_loads(query. load_spec)  # will load ony Foo.name and Bar.count

排序

from sqlalchemy_filters import apply_sort


# `query` should be a SQLAlchemy query object

sort_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 将尝试自动将模型与 query 联接，如果它们尚未存在并且提供了模型特定的排序。其行为与 apply_filters 相同。

这允许客户端在事先不指定所有可能的联接的情况下，对相关对象的字段进行排序。

分页

from sqlalchemy_filters import apply_pagination


# `query` should be a SQLAlchemy query object

query, pagination = apply_pagination(query, page_number=1, page_size=10)

page_size, page_number, num_pages, total_results = pagination

assert 10 == len(query)
assert 10 == page_size == pagination.page_size
assert 1 == page_number == pagination.page_number
assert 3 == num_pages == pagination.num_pages
assert 22 == total_results == pagination.total_results

过滤器格式

过滤器必须以列表的形式提供，并将依次应用。列表中的每个元素都将是一个字典，其格式如下

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

如果原始查询仅应用于一个模型，则“model”键是可选的。

如果只有一个过滤器，则可以省略包含它的列表。

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

其中field是要过滤的字段的名称，使用在op中提供的运算符（可选，默认为==）以及提供的value（可选，取决于运算符）进行过滤。

以下是可以使用的运算符列表

• is_null

• is_not_null

• ==，eq

• !=，ne

• >，gt

• <，lt

• >=，ge

• <=，le

• like

• ilike

• not_ilike

• in

• not_in

• any

• not_any

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'}
                ]
            },
        ],
    }
]

排序格式

排序元素必须以列表中的字典形式提供，并将依次应用。

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

其中field是要使用提供的direction进行排序的字段名称。

如果原始查询仅应用于一个模型，则“model”键是可选的。

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

运行测试

默认配置使用SQLite，MySQL（如果已安装驱动程序，则在使用tox时是这种情况）和PostgreSQL（如果已安装驱动程序，则在使用tox时是这种情况）来运行测试，以下URI

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'

将创建一个测试数据库，在测试期间使用，并在配置的每个RDBMS之后将其销毁。

存在用于在本地运行MySQL和PostgreSQL容器的Makefile目标，使用默认端口和配置

$ make mysql-container
$ make postgres-container

在本地运行测试

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

存在一些其他Makefile目标，可以用来运行测试

还有其他Makefile目标可以运行测试，但需要额外安装依赖项

$ 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

数据库管理系统

以下关系数据库管理系统（RDBMS）受支持（已测试）

• SQLite

• MySQL

• PostgreSQL

SQLAlchemy支持

以下SQLAlchemy版本受支持：1.0，1.1，1.2，1.3，1.4。

变更日志

请参阅CHANGELOG文档以了解每个版本的修复和增强。

许可

Apache 2.0。有关详细信息，请参阅LICENSE。