使用psycopg2向PostgreSQL jsonb字段写入数据遇格式错误求助

以下是我的代码：

from psycopg2.extras import Json
from psycopg2.extensions import register_adapter
from psycopg2.extras import execute_values
import psycopg2

register_adapter(dict, Json)

data = [{
    'end_of_epoch_data': ['GasCoin', [{'Input': 5}, {'Input': 6}, {'Input': 7}]],
}]

def get_upsert_sql(schema: str, table: str, columns: str, primary_keys: list | tuple | set):
    return f"""INSERT INTO {schema}.{table}
            ({', '.join(columns)}) VALUES %s
            ON CONFLICT ({','.join(primary_keys)}) DO UPDATE
            SET {', '.join([f"{col}=EXCLUDED.{col}" for col in columns if col not in primary_keys])}"""


def upsert(data: list, uri: str, schema: str, table: str, primary_keys: list | tuple | set):
    connection = psycopg2.connect(uri)
    cursor = connection.cursor()
    try:
        columns = data[0].keys()
        query = get_upsert_sql(schema, table, columns, primary_keys)
        values = [[d[col] for col in columns] for d in data]
        execute_values(cursor, query, values)
        connection.commit()
    except Exception as e:
        connection.rollback()
        raise e
    finally:
        cursor.close()
        connection.close()

执行时出现如下错误：

File "/Users/tests/test_pg_write.py", line 47, in upsert
    execute_values(cursor, query, values)
  File "/Users/venv/lib/python3.9/site-packages/psycopg2/extras.py", line 1299, in execute_values
    cur.execute(b''.join(parts))
psycopg2.errors.InvalidTextRepresentation: malformed array literal: "GasCoin"
LINE 2:             (end_of_epoch_data) VALUES (ARRAY['GasCoin',ARRA...
                                                      ^
DETAIL:  Array value must start with "{" or dimension information.

其中end_of_epoch_data是PostgreSQL表中的jsonb列，请问该如何解决？

更新
我发现错误原因似乎是尝试将Python列表写入jsonb字段，那使用json.dumps(data['end_of_epoch_data'])将列表转为字符串写入是否是正确的解决方案？

解决方案 #

错误原因分析 #

你仅注册了dict到Json适配器，但未处理Python列表类型。execute_values会默认把Python列表识别为PostgreSQL的数组类型，而非jsonb，这就导致PostgreSQL解析时出错——因为你的列表首元素是字符串"GasCoin"，不符合PostgreSQL数组需用{}包裹的语法要求。

正确解决方式 #

方式1：为列表注册Json适配器

直接将list类型也注册到Json适配器，让psycopg2自动把Python列表转为jsonb格式：

from psycopg2.extras import Json
from psycopg2.extensions import register_adapter

# 同时注册dict和list类型
register_adapter(dict, Json)
register_adapter(list, Json)

方式2：手动用Json包装目标字段

构造values时，针对jsonb字段的值手动用Json()包裹：

values = [[Json(d[col]) if col == 'end_of_epoch_data' else d[col] for col in columns] for d in data]

关于json.dumps的说明 #

用json.dumps转字符串写入不是最优方案：虽然能成功写入，但PostgreSQL会将其存储为普通字符串，而非jsonb类型。后续查询时需手动用jsonb()函数转换才能进行jsonb的专属操作（如->路径查询），完全失去了jsonb类型的优势，因此不推荐。

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