psycopg2 - insert None Using execute_values

Question

insert_bulk

def insert_bulk(products: list):
try:
sql = """
INSERT INTO product(
id, created_date_time, name, expiration_date
)
SELECT
id, NOW() AT TIME ZONE 'Asia/Seoul', name, expiration_date
FROM (VALUES %s)
AS products_insert (
id,
name,
expiration_date
);
"""
products_insert = [[value for value in product.values()] for product in products]
# products_insert
# [['1','apple', None],['2', 'banana', None],['3', 'meat', None], ...]
with Connector.connect() as connection:
with connection.cursor() as cursor:
psycopg2.extras.execute_values(
cursor,
sql,
products_insert,
page_size=500,
)
affected_row_count = cursor.rowcount
print(affected_row_count, "products inserted successfully")
except (Exception, Error) as error:
print(
"[product_repository] Error while insert_bulk",
error,
)

连接器

import psycopg2
from stock.properties.properties import properties
class Connector:
...
@staticmethod
def connect():
connection = psycopg2.connect(**properties.database)
return connection

当我尝试在expiration_date中插入None时，会发生以下错误。

列"expiration_date"的类型为date，但表达式的类型为text

提示：您需要重写或强制转换表达式。

我在更新中使用类型转换解决了这个问题
(expiration_date=product_update.expiration_date:：date(

sql = """
UPDATE product
SET
id = product_update.id,
name = product_update.name,
expiration_date = product_update.expiration_date::date
FROM (VALUES %s)
AS product_update (
id,
name,
expiration_date
) 
WHERE product_update.id = product.id;
"""

但这种方式在insert中不起作用。

那么如何在日期类型中插入None

Answer 1

模板参数是有效的

with Connector.connect() as connection:
with connection.cursor() as cursor:
psycopg2.extras.execute_values(
cursor,
sql,
products_insert,
template='(%s, %s, %s::date)'
page_size=500,
)