Python执行带参数PostgreSQL PL/pgSQL查询无更新问题排查

问题分析 #

核心问题是：PostgreSQL的PL/pgSQL匿名DO块在Python（SQLAlchemy）中执行时，参数绑定未正确生效，导致EXISTS条件始终不成立，只会走INSERT分支，但单独执行INSERT语句正常。

原因在于DO块是独立的PL/pgSQL执行单元，SQLAlchemy的命名参数（如:bid）无法直接穿透到DO块内部的逻辑中，重复使用的参数更是无法被正确解析，最终导致WHERE条件中的参数匹配失败，UPDATE分支永远不会触发。

解决方案 #

方案1：改用PostgreSQL UPSERT（推荐） #

PostgreSQL原生支持INSERT ... ON CONFLICT ... DO UPDATE语法，完全可以替代自定义DO块逻辑，更简洁且适配SQLAlchemy的参数绑定：

sql_query = text("""
    INSERT INTO data_check (bid, first_nid, first_datetime, second_nid, second_datetime, third_nid, third_datetime)
    VALUES (:bid, :first_nid, :first_datetime, :second_nid, :second_datetime, :third_nid, :third_datetime)
    ON CONFLICT (bid, first_nid) WHERE (first_datetime - EXCLUDED.first_datetime) BETWEEN '-00:30:00'::interval AND '00:30:00'::interval
    DO UPDATE SET
        second_nid = COALESCE(NULLIF(data_check.second_nid, ''), EXCLUDED.second_nid),
        second_datetime = COALESCE(data_check.second_datetime, EXCLUDED.second_datetime),
        third_nid = COALESCE(NULLIF(data_check.third_nid, ''), EXCLUDED.third_nid),
        third_datetime = COALESCE(data_check.third_datetime, EXCLUDED.third_datetime)
""")

kwargs['engine'].execute(sql_query, {
    "bid": str(row['bid']), 
    "first_nid": str(row['f_nid']), 
    "first_datetime": row['Date/Time'], 
    "second_nid": str(row['s_nid']), 
    "second_datetime": row['Date/Time_1'], 
    "third_nid": str(row['t_nid']), 
    "third_datetime": row['Date/Time_2']
})

注意：需要确保(bid, first_nid)是表的唯一约束（或主键），如果没有，先执行以下SQL创建：

ALTER TABLE data_check ADD CONSTRAINT unique_bid_firstnid UNIQUE (bid, first_nid);

方案2：修复DO块的参数传递 #

如果必须使用DO块，需要将参数显式传递到PL/pgSQL内部，通过声明变量来复用参数：

sql_query = text("""
DO $$
DECLARE
    p_bid text := :bid;
    p_first_nid text := :first_nid;
    p_first_datetime timestamp := :first_datetime;
    p_second_nid text := :second_nid;
    p_second_datetime timestamp := :second_datetime;
    p_third_nid text := :third_nid;
    p_third_datetime timestamp := :third_datetime;
BEGIN
    IF EXISTS (
        SELECT 1 FROM data_check 
        WHERE bid = p_bid 
          AND first_nid = p_first_nid
          AND (first_datetime - p_first_datetime) BETWEEN '-00:30:00'::interval AND '00:30:00'::interval
    ) THEN
        UPDATE data_check SET
            second_nid = COALESCE(NULLIF(second_nid, ''), p_second_nid),
            second_datetime = COALESCE(second_datetime, p_second_datetime),
            third_nid = COALESCE(NULLIF(third_nid, ''), p_third_nid),
            third_datetime = COALESCE(third_datetime, p_third_datetime)
        WHERE bid = p_bid 
          AND first_nid = p_first_nid
          AND (first_datetime - p_first_datetime) BETWEEN '-00:30:00'::interval AND '00:30:00'::interval;
    ELSE
        INSERT INTO data_check VALUES (p_bid, p_first_nid, p_first_datetime, p_second_nid, p_second_datetime, p_third_nid, p_third_datetime);
    END IF;
END $$;
""")

kwargs['engine'].execute(sql_query, {
    "bid": str(row['bid']), 
    "first_nid": str(row['f_nid']), 
    "first_datetime": row['Date/Time'], 
    "second_nid": str(row['s_nid']), 
    "second_datetime": row['Date/Time_1'], 
    "third_nid": str(row['t_nid']), 
    "third_datetime": row['Date/Time_2']
})

验证方法 #

开启PostgreSQL查询日志，查看实际执行的SQL语句，确认参数是否正确绑定：

1. 修改postgresql.conf，设置log_statement = 'all'
2. 重启PostgreSQL服务
3. 执行Python代码后，查看pg_log目录下的日志文件，检查参数是否被正确替换

内容的提问来源于stack exchange，提问作者RKIDEV