如何使用 sqlalchemy 在 postgres JSONB 中查找嵌套值?
How to find nested value in postgres JSONB using sqlalchemy?
我有一个带有 JSONB 列名称 entity
的 postgres table。 json结构的例子是:
{
"some_key": "some value",
"properties": {
"name": ["name_value"],
}
}
我需要按 name_value
查找记录。我可以通过查询得到它:
SELECT entity FROM my_table where entity->'properties' @> {"name": ["name_value"]};
问题是:我找不到使用 sqlalchemy ORM 创建相同查询的方法。
更新:
我的模型是动态的,因为多个 table 使用相同的结构只是不同的 table 名称。这是代码:
...
Base = declarative_base(engine)
class ForeignBase(AbstractConcreteBase, Base):
pass
def make_model(name):
class Entity(ForeignBase):
__tablename__ = name
__mapper_args__ = {"polymorphic_identity": name, "concrete": True}
__table_args__ = {"extend_existing": True}
id = Column(String, primary_key=True)
entity = Column(JSONB, nullable=True)
...
configure_mappers()
return Entity
然后我使用 3 个函数来启动我的模型并得到一个我当前需要的函数:
def get_model_list():
tables_names_list = get_table_name_list()
model_list = [make_model(PREFIX + table_name) for table_name in tables_names_list]
return model_list
def get_tables_dict():
return {table.__tablename__: table for table in ForeignBase.__subclasses__()}
def get_table_object(table_name):
return get_tables_dict().get(table_name)
您可以使用以下 SQLAlchemy 表达式。对于 JSONB 列,运算符 @> 在 SQLAlchemy 中是 contains()
Session().query(MyTable).filter(MyTable.entity["properties"].contains({"name": ["name_value"]})).with_entities(MyTable.entity).all()
我有一个带有 JSONB 列名称 entity
的 postgres table。 json结构的例子是:
{
"some_key": "some value",
"properties": {
"name": ["name_value"],
}
}
我需要按 name_value
查找记录。我可以通过查询得到它:
SELECT entity FROM my_table where entity->'properties' @> {"name": ["name_value"]};
问题是:我找不到使用 sqlalchemy ORM 创建相同查询的方法。
更新: 我的模型是动态的,因为多个 table 使用相同的结构只是不同的 table 名称。这是代码:
...
Base = declarative_base(engine)
class ForeignBase(AbstractConcreteBase, Base):
pass
def make_model(name):
class Entity(ForeignBase):
__tablename__ = name
__mapper_args__ = {"polymorphic_identity": name, "concrete": True}
__table_args__ = {"extend_existing": True}
id = Column(String, primary_key=True)
entity = Column(JSONB, nullable=True)
...
configure_mappers()
return Entity
然后我使用 3 个函数来启动我的模型并得到一个我当前需要的函数:
def get_model_list():
tables_names_list = get_table_name_list()
model_list = [make_model(PREFIX + table_name) for table_name in tables_names_list]
return model_list
def get_tables_dict():
return {table.__tablename__: table for table in ForeignBase.__subclasses__()}
def get_table_object(table_name):
return get_tables_dict().get(table_name)
您可以使用以下 SQLAlchemy 表达式。对于 JSONB 列,运算符 @> 在 SQLAlchemy 中是 contains()
Session().query(MyTable).filter(MyTable.entity["properties"].contains({"name": ["name_value"]})).with_entities(MyTable.entity).all()