Sqlalchemy 按连续列表中的单个值过滤

Question

基本上，我有一个 SQLAlchemy 模型，其中类别存储为 json 列表。该列表可能类似于[1, 6, 7]。我需要知道如何（以及是否）可以以某种方式按此列表中的值进行过滤。 Object.query.filter_by(tag in json.loads(category).all() 之类的东西（json.loads() 会在这里返回一个 python 列表，而不是一个字典）。

def sort_objects(mode, return_size, **kwargs):
    if "page" in kwargs.keys():
        page = kwargs['page']
    else: page = 1
    if "tag" in kwargs.keys():
        tag = kwargs['tag']
    else: tag = 0

    if mode.lower() == "new":
        q = Object.query.filter(tag in json.loads(category)).\
        order_by(Object.uploaded_utc.desc()).paginate(page, return_size, False).items```

Answer 1

为此，您可以在 PostgreSQL 中的 JSONB 列上使用 ? 运算符（或 @> 用于整数值）。见the JSON page for a detailed reference

正如你提到的“标签”，我假设数组中的值是字符串，所以我们可以使用 ? 运算符。

要解决这个难题，需要知道一些事情：

值需要在 postgres JSONB 类型中

正如评论中提到的，这些值目前没有存储为JSONB，而是TEXT，所以我们需要转换它们。这可以使用sqlalchemy.cast 来完成。我们还需要为该演员从sqlalchemy.dialects.postgresql 拉入JSONB 类型。

我们需要调用一个非标准的算子

完成此操作的方法是在 SQLAlchemy 模型的列上调用 .op()。这将返回一个新的“可调用”，它采用该运算符的“右手边”（其中“左手边”是表格列）。因此，例如，如果我们有一个名为 mycolumn 的列和一个名为 yoink 的运算符，我们想要执行 mycolumn yoink 10，我们需要编写以下内容：

    Table.mycolumn.op("yoink")(10)
    \------------------------/\--/
          mycolumn yoink       10

然后可以将所有这些组合到以下过滤器表达式（使用“needle”作为我们搜索的标签值：

session.query(MyTable).filter(
    cast(MyTable.myjsoncolumn, JSON).op("?")(needle)
)

为您的代码

鉴于您的问题，我的猜测是您的代码的查询是这样的：

session.query(MyTable).filter(
    cast(Object.category, JSON).op("?")(tag)
)

关于 TEXT 与 JSONB 的注意事项

如果您的“类别”列包含 JSON 数据，那么在内部使用 JSONB 类型而不是纯文本会很有意义。如果需要，它可以让你做更多花哨的查询（比如这个问题中的那个），而不需要强制转换。

完全运行的示例

这是一个示例，它创建一个带有 JSON 列的简单表，然后插入一些虚拟数据并在最后进行查询。

我使用一个用于 postgres 的隔离 docker 容器运行此程序：

docker run --rm --name pgdb -e POSTGRES_PASSWORD=foobar -p 5432:5432 postgres

from json import dumps

from sqlalchemy import Column, Integer, String, cast, create_engine
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()


class SomeClass(Base):
    __tablename__ = "some_table"
    id = Column(Integer, primary_key=True)
    data = Column(String())


engine = create_engine("postgresql://postgres:foobar@127.0.0.2/postgres")
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Ensure we are working with an empty table on repeating calls
session.query(SomeClass).delete()
session.add_all(
    [
        SomeClass(data=dumps(["tag-10", "tag-20", "tag-30"])),
        SomeClass(data=dumps(["tag-10", "tag-20", "tag-30"])),
        SomeClass(data=dumps(["tag-10", "tag-20", "tag-30"])),
        SomeClass(data=dumps(["tag-10", "tag-20", "tag-30"])),
        SomeClass(data=dumps(["tag-1", "tag-2", "tag-3"])),
        SomeClass(data=dumps(["tag-2", "tag-3", "tag-4"])),
        SomeClass(data=dumps(["tag-3", "tag-1", "tag-2"])),
        SomeClass(data=dumps(["tag-2", "tag-3", "tag-1"])),
    ]
)
session.commit()

print("--- Unfiltered:")
q = session.query(SomeClass)
for row in q:
    print(row, row.data)

print("--- Filtered:")
q = session.query(SomeClass).filter(
    cast(SomeClass.data, JSONB).op("?")('tag-2')
)
for row in q:
    print(row, row.data)