如何从sqlalchemy中的jsonb列的嵌套列表中返回特定的字典键

Question

我试图从用PostgreSQL存储的jsonb数据集中返回一些命名列。你知道吗

我能够直接运行满足我需要的原始查询，但是我尝试使用SQLAlchemy运行查询，以确保我的代码是“pythonic”的并且易于阅读。你知道吗

返回正确结果（两列）的查询是：

SELECT  
    tmp.item->>'id',
    tmp.item->>'name'
FROM (SELECT jsonb_array_elements(t.data -> 'users') AS item FROM tpeople t) as tmp

示例json（每个用户有20多列）

{ "results":247, "users": [
{"id":"202","regdate":"2015-12-01","name":"Bob Testing"},
{"id":"87","regdate":"2014-12-12","name":"Sally Testing"},
{"id":"811", etc etc}
...
]}

这个表非常简单，有一个PK、json提取的datetime和提取的jsonb列

CREATE TABLE tpeople
(
    record_id bigint NOT NULL DEFAULT nextval('"tpeople_record_id_seq"'::regclass) ( INCREMENT 1 START 1 MINVALUE 1 MAXVALUE 9223372036854775807 CACHE 1 ),
    scrape_time timestamp without time zone NOT NULL,
    data jsonb NOT NULL,
    CONSTRAINT "tpeople_pkey" PRIMARY KEY (record_id)
);

此外，我还有一个People类，如下所示：

class people(Base):
    __tablename__ = 'tpeople'

    record_id = Column(BigInteger, primary_key=True, server_default=text("nextval('\"tpeople_record_id_seq\"'::regclass)"))
    scrape_time = Column(DateTime, nullable=False)
    data = Column(JSONB(astext_type=Text()), nullable=False)

目前，返回这两列的代码如下所示：

from db.db_conn import get_session // Generic connector for my db
from model.models import people
from sqlalchemy import func,

sess = get_session()

sub = sess.query(func.jsonb_array_elements(people.data["users"]).label("item")).subquery()
test = sess.query(sub.c.item).select_entity_from(sub).all()

SQLAlchemy生成以下SQL：

SELECT anon_1.item AS anon_1_item 
FROM (SELECT jsonb_array_elements(tpeople.data -> %(data_1)s) AS item 
FROM tpeople) AS anon_1
{'data_1': 'users'}

但是，我似乎做的任何事情都不允许我只获取项本身中的某些列，就像我可以编写的原始SQL一样。我尝试过的一些方法如下（它们都出错了）：

test = sess.query("sub.item.id").select_entity_from(sub).all()

test = sess.query(sub.item.["id"]).select_entity_from(sub).all()

aas = func.jsonb_to_recordset(people.data["users"])
res = sess.query("id").select_from(aas).all()

sub = select(func.jsonb_array_elements(people.data["users"]).label("item"))

目前，我可以在一个简单的for循环中提取所需的列，但这似乎是一种很难做到的方法，而且我确信我遗漏了一些非常明显的东西。你知道吗

for row in test:
    print(row.item['id'])

Answer 1

搜索了几个小时后，终于发现一些人在试图得到另一个结果时意外地做了这件事。你知道吗

sub = sess.query(func.jsonb_array_elements(people.data["users"]).label("item")).subquery()
tmp = sub.c.item.op('->>')('id')
tmp2 = sub.c.item.op('->>')('name')
test = sess.query(tmp, tmp2).all()