我有一個看起來像的模式Column | Type |-------------------------------------------------------message_id | integer | user_id | integer | body | text | created_at | timestamp without time zone | source | jsonb | symbols | jsonb[] |我正在嘗試使用 psycopg2 通過 psycopg2.Cursor.copy_from() 插入數(shù)據(jù)，但我遇到了很多問題，試圖弄清楚應(yīng)該如何格式化 jsonb[] 對象。當(dāng)我做一個 JSON 對象的直接列表時，我得到一個錯誤，看起來像psycopg2.errors.InvalidTextRepresentation: malformed array literal: "[{'id': 13016, 'symbol':.... DETAIL: "[" must introduce explicitly-specified array dimensions.我在雙引號和大括號上嘗試過許多不同的轉(zhuǎn)義。如果我對我的數(shù)據(jù)執(zhí)行 json.dumps()，我會收到以下錯誤。psycopg2.errors.InvalidTextRepresentation: invalid input syntax for type jsonDETAIL: Token "'" is invalid.從此代碼段收到此錯誤messageData = []symbols = messageObject["symbols"]newSymbols = []for symbol in symbols: toAppend = symbol toAppend = refineJSON(json.dumps(symbol)) toAppend = re.sub("{", "\{", toAppend) toAppend = re.sub("}", "\}", toAppend) toAppend = re.sub('"', '\\"', toAppend) newSymbols.append(toAppend)messageData.append(set(newSymbols))我也愿意將列定義為不同的類型（例如，文本），然后嘗試進(jìn)行轉(zhuǎn)換，但我也無法做到這一點。messageData 是調(diào)用 psycopg2.Cursor.copy_from() 的輔助函數(shù)的輸入def copy_string_iterator_messages(connection, messages, size: int = 8192) -> None: with connection.cursor() as cursor: messages_string_iterator = StringIteratorIO(( '|'.join(map(clean_csv_value, (messageData[0], messageData[1], messageData[2], messageData[3], messageData[4], messageData[5], messageData[6], messageData[7], messageData[8], messageData[9], messageData[10], messageData[11], ))) + '\n' for messageData in messages )) # pp.pprint(messages_string_iterator.read()) cursor.copy_from(messages_string_iterator, 'test', sep='|', size=size) connection.commit()
1 回答
三國紛爭
TA貢獻(xiàn)1804條經(jīng)驗 獲得超7個贊
你的問題讓我很好奇。下面這個對我有用。我懷疑是否可以解決轉(zhuǎn)義到 CSV 或從 CSV 轉(zhuǎn)義的問題。
我的表：
=# \d jbarray
Table "public.jbarray"
Column | Type | Collation | Nullable | Default
---------+---------+-----------+----------+-------------------------------------
id | integer | | not null | nextval('jbarray_id_seq'::regclass)
symbols | jsonb[] | | |
Indexes:
"jbarray_pkey" PRIMARY KEY, btree (id)
完全獨立的 Python 代碼：
mport json
import psycopg2
con = psycopg2.connect('dbname=<my database>')
some_objects = [{'id': x, 'array': [x, x+1, x+2, {'inside': x+3}]} for x in range(5)]
insert_array = [json.dumps(x) for x in some_objects]
print(insert_array)
c = con.cursor()
c.execute("insert into jbarray (symbols) values (%s::jsonb[])", (insert_array,))
con.commit()
結(jié)果：
=# select * from jbarray;
-[ RECORD 1 ]-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
id | 1
symbols | {"{\"id\": 0, \"array\": [0, 1, 2, {\"inside\": 3}]}","{\"id\": 1, \"array\": [1, 2, 3, {\"inside\": 4}]}","{\"id\": 2, \"array\": [2, 3, 4, {\"inside\": 5}]}","{\"id\": 3, \"array\": [3, 4, 5, {\"inside\": 6}]}","{\"id\": 4, \"array\": [4, 5, 6, {\"inside\": 7}]}"}
=# select id, unnest(symbols) from jbarray;
-[ RECORD 1 ]----------------------------------------
id | 1
unnest | {"id": 0, "array": [0, 1, 2, {"inside": 3}]}
-[ RECORD 2 ]----------------------------------------
id | 1
unnest | {"id": 1, "array": [1, 2, 3, {"inside": 4}]}
-[ RECORD 3 ]----------------------------------------
id | 1
unnest | {"id": 2, "array": [2, 3, 4, {"inside": 5}]}
-[ RECORD 4 ]----------------------------------------
id | 1
unnest | {"id": 3, "array": [3, 4, 5, {"inside": 6}]}
-[ RECORD 5 ]----------------------------------------
id | 1
unnest | {"id": 4, "array": [4, 5, 6, {"inside": 7}]}
如果插入性能對您來說太慢，那么您可以按照此處的說明prepared statement使用with 。我用過那個組合，速度非常快。execute_batch()
添加回答
舉報
0/150
提交
取消