使用on_conflict_do_update执行Upsert时添加updated_at列报错求助

在使用on_conflict_do_update执行Upsert操作时，尝试在更新内容中添加timestamp类型的updated_at列，运行代码最后一行时触发如下错误：

已发生异常：ProgrammingError
(psycopg2.errors.AmbiguousColumn) column reference "updated_at" is ambiguous
LINE 1: ...r, properties = excluded.properties, updated_at = updated_at

当前代码实现：

def postgres_upsert(table, conn, keys, data_iter):
    from sqlalchemy.dialects.postgresql import insert

    data = [dict(zip(keys, row)) for row in data_iter]

    insert_statement = insert(table.table).values(data)

    x = {c.key: c for c in insert_statement.excluded}
    x["updated_at"] = Column('updated_at', TIMESTAMP, default=datetime.datetime.now())

    upsert_statement = insert_statement.on_conflict_do_update(
        constraint=f"{table.table.name}_pkey",
        # set_={c.key: c for c in insert_statement.excluded},
        set_=x,
    )
    conn.execute(upsert_statement)

错误根源是直接用Column对象赋值updated_at，导致生成的SQL语句中updated_at = updated_at出现歧义——数据库无法区分左右两侧的updated_at分别来自原表还是excluded虚拟表。正确做法是直接指定当前时间值或调用数据库内置函数，同时确保表中已存在updated_at列。

修改后的代码：

from sqlalchemy import func
from sqlalchemy.dialects.postgresql import insert
from datetime import datetime

def postgres_upsert(table, conn, keys, data_iter):
    data = [dict(zip(keys, row)) for row in data_iter]

    insert_statement = insert(table.table).values(data)

    update_dict = {c.key: c for c in insert_statement.excluded}
    # 推荐用数据库端函数生成时间，避免Python与数据库的时间差
    update_dict["updated_at"] = func.now()
    # 也可以用Python本地时间：update_dict["updated_at"] = datetime.now()

    upsert_statement = insert_statement.on_conflict_do_update(
        constraint=f"{table.table.name}_pkey",
        set_=update_dict,
    )
    conn.execute(upsert_statement)

关键说明 #

• 禁止用Column对象赋值更新字段，直接传入明确的时间值或数据库函数即可消除列引用歧义
• 使用func.now()让PostgreSQL自行生成当前时间，比Python本地时间更可靠，能避免跨时区或时钟不一致问题
• 需确保目标数据库表已创建updated_at列，类型为TIMESTAMP或TIMESTAMP WITH TIME ZONE

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