SqlAlchemy - 如何根据保存为 NestedMutableJson 的键属性查询表

Question

假设一个 Postgres 用户表包含一个 NestedMutableJson 类型的属性。

first_name character varying (120)
last_name character varying (120
country_info NestedMutableJson
...
from sqlalchemy_json import NestedMutableJson
country_info = db.Column(NestedMutableJson, nullable=True)

country_info = {"name": "UK", "code": "11"}

如何根据 country_info 键查询用户表。

POSTGRES 查询

SELECT * FROM user WHERE country_info ->> 'name' = 'UK'

是否有任何 SqlAlchemy 方式给出相同的查询结果？

我尝试了几种方法，例如：

方式一：

User.query.filter(User.country_info['name'].astext == 'UK').all()

错误：

Operator 'getitem' is not supported on this expression

方式二：

User.query.filter(User.country_info.op('->>')('name') == 'UK').all()

问题：

Always getting an empty response 

我想知道是不是列定义db.Column(NestedMutableJson, nullable=True)引起的问题

我避免使用 db.session.execute("SELECT * FROM user WHERE country_info ->> 'name' = 'UK'").fetchall()。寻找别的东西

Answer 1

只需使用 text，它允许您在 orm 进程中编写纯文本过滤器，它的工作原理类似于 的 get 函数dict，它也可以处理None字段。

User.query.filter(text("COALESCE(user.country_info ->> 'name', '') = 'UK'")).all()

注意表的名字应该是数据库中的真实名字

Answer 2

查看JSON 数据类型文档。

你应该可以使用这个过滤子句 select(CountryInfo).filter(CountryInfo.country_info['name'].astext == "UK")

Answer 3

你可以使用.op('->>')来使用PostgreSQL操作符->>，方式如下：

from sqlalchemy import Column, Integer, create_engine, String, select
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
from sqlalchemy.dialects.postgresql import JSON
from sqlalchemy_json import NestedMutableJson

dburl = 'postgresql://...'
Base = declarative_base()

class CountryInfo(Base):
    __tablename__ = 'country_info'
    id = Column(Integer, unique=True, nullable = False, primary_key=True)
    name = Column(String)
    country_info = Column(NestedMutableJson, nullable=True)

    def __repr__(self):
        return f'CountryInfo({self.name!r}, {self.country_info!r})'

engine = create_engine(dburl, future=True, echo=True)

Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

with Session(engine) as session:
    test = CountryInfo(name='test', country_info={"name": "UK", "code": "11"})
    test2 = CountryInfo(name='test2', country_info={"name": "NL", "code": "12"})
    test3 = CountryInfo(name='test3', country_info={"name": "UK", "code": "13"})

    session.add(test)
    session.add(test2)
    session.add(test3)
    session.commit()

    stmt = select(CountryInfo).filter(CountryInfo.country_info.op('->>')('name') == "UK")

    query = session.execute(stmt).all()
    for row in query:
        print(row)

这会产生以下 SQL：

SELECT country_info.id, country_info.name, country_info.country_info
FROM country_info
WHERE (country_info.country_info ->> %(country_info_1)s) = %(param_1)s

{'country_info_1': 'name', 'param_1': 'UK'}

结果：

(CountryInfo('test', {'name': 'UK', 'code': '11'}),)
(CountryInfo('test3', {'name': 'UK', 'code': '13'}),)