如何在SQLAlchemy中从PostgreSQL的jsonb列嵌套列表返回指定键

阿华AIGC实验室

2026-5-12

用SQLAlchemy实现PostgreSQL JSONB数组展开与字段提取

我来帮你搞定这个SQLAlchemy处理JSONB的问题！你的原生SQL逻辑很清晰，我们可以把它完美转换成Python风格的易读代码，不管是用ORM模型还是SQLAlchemy Core都能实现，完全不用循环提取字段。

一、基于ORM模型的实现（推荐，符合你的Python风格需求）

假设你的People模型类定义如下（对应tpeople表）：

from sqlalchemy import Column, Integer, DateTime, JSONB
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class People(Base):
    __tablename__ = 'tpeople'
    record_id = Column(Integer, primary_key=True)
    scrape_time = Column(DateTime)
    data = Column(JSONB)

我们可以分两步实现和原生SQL完全等价的查询：

1. 构建子查询：展开JSONB数组

首先用func.jsonb_array_elements把data->'users'数组展开成每行一个item，对应原生SQL里的子查询tmp：

from sqlalchemy import func, select
from sqlalchemy.orm import Session

# 假设你已经创建了Session实例session
subquery = (
    select(func.jsonb_array_elements(People.data['users']).label('item'))
    .select_from(People)
).subquery('tmp')

2. 主查询：提取指定字段

从子查询中提取item里的id和name，用.astext对应原生SQL的->>操作符（把JSONB值转为文本）：

query = select(
    subquery.c.item['id'].astext.label('user_id'),
    subquery.c.item['name'].astext.label('user_name')
)

# 执行查询并获取结果
results = session.execute(query).all()

结果处理

你可以直接把结果转换成Python字典列表，非常方便：

user_list = [{'id': row.user_id, 'name': row.user_name} for row in results]

二、基于SQLAlchemy Core的实现（如果不用ORM模型）

如果直接操作表而不是ORM模型，代码逻辑类似：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, DateTime, JSONB
from sqlalchemy import func, select

# 初始化连接和表对象
engine = create_engine('postgresql://your_user:your_pass@your_host/your_db')
metadata = MetaData()
tpeople = Table(
    'tpeople', metadata,
    Column('record_id', Integer, primary_key=True),
    Column('scrape_time', DateTime),
    Column('data', JSONB)
)

# 子查询展开数组
subquery = (
    select(func.jsonb_array_elements(tpeople.c.data['users']).label('item'))
    .select_from(tpeople)
).subquery('tmp')

# 主查询提取字段
query = select(
    subquery.c.item['id'].astext.label('user_id'),
    subquery.c.item['name'].astext.label('user_name')
)

# 执行查询
with engine.connect() as conn:
    results = conn.execute(query).fetchall()

关键知识点解析

func.jsonb_array_elements：对应PostgreSQL原生的jsonb_array_elements函数，专门用来展开JSONB数组为行数据
data['users']：SQLAlchemy中访问JSONB字段的子属性，等价于原生SQL的data -> 'users'
.astext：把JSONB类型的值转换为文本类型，等价于原生SQL的->>操作符；如果需要保留JSON类型，去掉.astext即可（对应->操作符）
子查询的列引用：子查询创建后，通过subquery.c.xxx来访问子查询中的列，这里就是subquery.c.item