How to use DO $$ block with psycopg3?

Question

Context

I have a table that is using temporal versioning where the "active" row (for a retailer) is where expire_tstamp is null.

Problem

I want to execute once to do an "upsert".

1. If there are no existing active rows for a retailer, insert.
2. If there are existing active rows for a retailer, expire them, then insert.

But I keep getting error:

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

What I've tried so far

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_xml(
    aurora_cursor: PGCursor, retailer_id: RetailerId | ClientId, xml: str
) -> None:
    params = {"retailer_id": retailer_id, "xml": xml}

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

My table

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);

I've tried the following but is getting unique constraint error

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

An 'answer' on what I think is going on:

import psycopg
from psycopg import sql, ClientCursor

con = psycopg.connect("postgresql://postgres:postgres@127.0.0.1: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')]

psycopg3 differences. In psycopg(3) the manner in which parameters are bound to the query changed. That now happens on the server not the client as before. This shows up as an error in the first example above. In the second example ClientCursor is used. This emulates the psycopg2 process of client side parameter binding and allow the statement to complete successfully. The last example uses the sql to dynamically create the SQL statement with the supplied data.