将 Python 转义为 `json[]` 以进行 `COPY FROM` PostgreSQL 插入？

Question

如何正确转义以插入

json[]

列？

目前我得到这个错误：

psycopg2.errors.InvalidTextRepresentation: malformed array literal: "{"{"jj":null,"text":"bop"}",}"

在内部，它被转义为 Python 字符串，内容为：

{"{\"jj\":null,\"text\":\"bop\"}",}

。还需要转义吗？

---

from io import StringIO

def psql_insert_copy(table, conn, keys, data_iter):
    with conn.cursor() as cur:
        s_buf = StringIO()
        s_buf.writelines(
            "\n".join(
                map(lambda line: "\t".join(map(str, map(parse_col, line))), data_iter)
            )
        )
        s_buf.seek(0)

        columns = ", ".join(keys[1:] if keys and keys[0] == "index" else keys)
        if table.schema:
            table_name = '{}."{}"'.format(table.schema, table.name)
        else:
            table_name = table.name

        sql = "COPY \"{}\" ({}) FROM STDIN WITH null as 'null'".format(
            table_name, columns
        )
        cur.copy_expert(sql=sql, file=s_buf)

我的辅助函数：

import numpy as np
from json import dumps

def parse_col(col):
    if isinstance(col, np.ndarray):
        return parse_col(col.tolist()) if col.size > 0 else "null"
    elif isinstance(col, bool):
        return int(col)
    elif isinstance(col, bytes):
        return parse_col(col.decode("utf8"))
    elif isinstance(col, (complex, int)):
        return col
    elif isinstance(col, float):
        return int(col) if col.is_integer() else col
    elif col in (None, "{}", "[]") or not col:
        return "null"
    elif isinstance(col, str):
        return {"True": 1, "False": 0}.get(col, col)  # '"{}"'.format(col)
    elif isinstance(col, (list, tuple, set, frozenset)):
        return "{{{0}{1}}}".format(
            ",".join(
                map(  # partial(maybe_quote_and_escape, ch="'")
                    '"{}"'.format, map(parse_col, col)
                )
            ),
            "," if len(col) == 1 else "",
        )
    elif isinstance(col, dict):
        return dumps(col, separators=(",", ":")).replace('"', '\\"')
    elif isinstance(col, datetime):
        return col.isoformat()
    else:
        raise NotImplementedError(type(col))

用法：

from itertools import repeat
import psycopg2

conn = psycopg2.connect("dbname=test user=postgres")
conn.cursor().execute(
    "CREATE TABLE my_table ("
    "  json_arr_col json[],"
    "  id integer generated by default as identity primary key"
    ");")

psql_insert_copy(
    conn=conn,
    keys=("json_arr_col", "id"),
    data_iter=repeat(({"jj": None, "text": "bop"},),5),
    table="my_table"
)

Answer 1

一个简短的例子：

cat js_vals.json 
[{"one": 1, "two": 2, "three": 3}]
[{"dog": "ranger", "cat": "cassie", "fish": "trout"}]


import psycopg2
from psycopg2 import sql

con = psycopg2.connect(dbname="test", host='localhost', user='postgres')
cur = con.cursor()
cur.execute("""CREATE TABLE my_table (
    json_arr_col jsonb,
    id integer generated by default as identity primary key)""")
con.commit()

# sql.Identifier("public", "my_table") becomes "public"."my_table".
sql = sql.SQL("COPY {} ({}) FROM STDIN").format(sql.Identifier("public", "my_table"), sql.Identifier("json_arr_col"))

with open("js_vals.json", "r") as js_file:
    cur.copy_expert(sql, js_file)
con.commit()

cur.execute("select * from my_table")
cur.fetchall()
[([{'one': 1, 'two': 2, 'three': 3}], 1),
 ([{'cat': 'cassie', 'dog': 'ranger', 'fish': 'trout'}], 2)]