使用asyncpg调用PostgreSQL的jsonb_agg()返回前缀问题排查

你遇到的情况是asyncpg的正常行为：fetch()方法默认返回Record对象的列表，因为你的SQL查询通过jsonb_agg聚合后只返回一行一列的结果，所以result_raw是包含1个Record的列表，需要通过键名提取对应的值。另外你代码里的json.loads()完全多余——asyncpg会自动把PostgreSQL的jsonb类型转换成Python原生的列表/字典对象，不用手动解析JSON字符串。

下面是两种直接获取目标数据的方案：

方案1：改用fetchrow()获取单行结果 #

既然你的查询只会返回一行数据，用fetchrow()代替fetch()可以直接拿到单个Record对象，再提取字段值就是你需要的列表：

修改fetch_data_from_db()函数：

async def fetch_data_from_db():
    """get data from DB."""
    conn = await asyncpg.connect(DSN)
    result = await conn.fetchrow(QUERY)  # 改用fetchrow()获取单行
    await conn.close()
    return result["jsonb_agg"]  # 直接返回聚合后的Python列表

对应的get_list()和main()可以简化：

async def get_list():
    """Get List"""
    result = await fetch_data_from_db()
    return result

async def main():
    json_result = await get_list()  # 直接拿到原生Python列表

    print(f"type and len of json_result: {type(json_result)}/{len(json_result)}")
    try:
        for persona_data in json_result:
            persona = Person(**persona_data)
            print("Valid JSON object:", persona)
    except ValueError as e:
        print(e)

方案2：在fetch后直接提取目标字段 #

如果坚持使用fetch()，可以在获取结果后直接提取第一个Record的jsonb_agg值：

修改fetch_data_from_db()：

async def fetch_data_from_db():
    """get data from DB."""
    conn = await asyncpg.connect(DSN)
    results = await conn.fetch(QUERY)
    await conn.close()
    return results[0]["jsonb_agg"]  # 直接返回聚合后的列表

这样后续代码无需再做额外处理，get_list()返回的就是原生Python列表。

修改后的输出 #

运行调整后的代码，会得到如下输出：

type and len of json_result: <class 'list'>/3
Valid JSON object: id=8 name='Donald' surname='Duck'
Valid JSON object: id=9 name='Daisy' surname='Duck'
Valid JSON object: id=10 name='Scrooge' surname='McDuck'

内容的提问来源于stack exchange，提问作者Roby