如何将 DO $$ 块与 psycopg3 一起使用？

Question

背景

我有一个使用临时版本控制的表，其中“活动”行（对于零售商）是 expire_tstamp 为空的位置。

问题

我想执行一次来进行“更新插入”。

1. 如果零售商没有现有的活动行，请插入。
2. 如果零售商存在现有的活动行，则使它们过期，然后插入。

但我不断收到错误：

psycopg.errors.IndeterminateDatatype: could not determine data type of parameter $2

到目前为止我已经尝试过的事情

QUERY = """
DO $$
BEGIN
    IF EXISTS (
        SELECT 1 
        FROM retailer_xml 
        WHERE retailer_id = %(retailer_id)s::integer AND expire_tstamp IS NULL
    ) THEN
        UPDATE retailer_xml
        SET expire_tstamp = CURRENT_TIMESTAMP
        WHERE retailer_id = %(retailer_id)s::integer AND expire_tstamp IS NULL;
    END IF;

    INSERT INTO retailer_xml (retailer_id, xml)
    VALUES (%(retailer_id)s::integer, %(xml)s::text);
END $$;
"""

def insert_retailer_department_remap_list(
    aurora_cursor: PGCursor, retailer_id: RetailerId | ClientId, xml: str
) -> None:
    params = {"retailer_id": retailer_id, "xml": xml}

    aurora_cursor.execute(
        query=QUERY,
        params=params,
    )

我的桌子

CREATE TABLE public.retailer_xml (
    retailer_xml_id bigserial NOT NULL,
    retailer_id int4 NOT NULL,
    "xml" text NOT NULL,
    create_tstamp timestamptz DEFAULT now() NOT NULL,
    modify_tstamp timestamptz DEFAULT now() NOT NULL,
    expire_tstamp timestamptz NULL,
    CONSTRAINT retailer_xml_pkey PRIMARY KEY (retailer_xml_id)
);
CREATE UNIQUE INDEX retailer_xml_retailer_id_idx ON public.retailer_xml USING btree (retailer_id) WHERE (expire_tstamp IS NULL);

我尝试了以下方法，但遇到了唯一约束错误

WITH expired AS (
    UPDATE remap_list_xml
    SET expire_tstamp = CURRENT_TIMESTAMP
    WHERE retailer_id = %s AND expire_tstamp IS NULL
    RETURNING remap_list_xml_id
)
INSERT INTO remap_list_xml (retailer_id, xml)
VALUES (%s, %s);

Answer 1

import psycopg
from psycopg import sql, ClientCursor

con = psycopg.connect("postgresql://postgres:[email protected]:5432/test")

createQry = "create table psycopg_test(id integer, fld_1 varchar)"
doQry = """DO $$
BEGIN
    INSERT INTO psycopg_test values(%(id)s, %(fld_1)s);
END;
$$
"""

cur.execute(createQry)
con.commit()

cur.execute(doQry, {"id": 1, "fld_1": "test"})

IndeterminateDatatype: could not determine data type of parameter $2

con.rollback()
client_cur = ClientCursor(con)
client_cur.execute(doQry, {"id": 1, "fld_1": "test"})
cur.execute("select * from psycopg_test")
cur.fetchall()
[(1, 'test')]

doQrySQL = sql.SQL("""DO $$
BEGIN
    INSERT INTO psycopg_test values({}, {});
END;
$$
""").format(sql.Literal(2), sql.Literal('test2') )
cur.execute(doQrySQL)
con.commit()
cur.fetchall()
[(1, 'test'), (2, 'test2')]