如何在SQLAlchemy的insert_on_conflict_nothing中自动获取主键？

错误原因 #

你在自定义方法中直接访问的table是Pandas内部的SQLTable实例，而非SQLAlchemy的Table对象，所以它没有primary_key属性。必须获取对应的SQLAlchemy Table实例才能读取主键信息。

方法一：直接调用SQLTable内部的SQLAlchemy Table #

修改自定义方法，访问SQLTable的table属性（该属性对应SQLAlchemy的Table对象）来获取主键：

def insert_on_conflict_nothing(table, conn, keys, data_iter):
    data = [dict(zip(keys, row)) for row in data_iter]
    # 获取SQLAlchemy Table实例
    sa_table = table.table
    # 提取主键列名
    pkey = [pk_col.name for pk_col in sa_table.primary_key.columns.values()]
    stmt = insert(sa_table).values(data).on_conflict_do_nothing(index_elements=pkey)
    result = conn.execute(stmt)
    return result.rowcount

方法二：手动加载SQLAlchemy Table（更可靠） #

如果方法一出现表结构未加载的问题，可以在自定义方法内主动从数据库加载表结构，确保能正确获取主键：

from sqlalchemy import Table, MetaData

def insert_on_conflict_nothing(table, conn, keys, data_iter):
    data = [dict(zip(keys, row)) for row in data_iter]
    # 从数据库加载表结构
    meta = MetaData()
    sa_table = Table(
        table.name,
        meta,
        schema=table.schema,
        autoload_with=conn.engine
    )
    # 提取主键列名
    pkey = [pk_col.name for pk_col in sa_table.primary_key.columns.values()]
    stmt = insert(sa_table).values(data).on_conflict_do_nothing(index_elements=pkey)
    result = conn.execute(stmt)
    return result.rowcount

注意事项 #

• 确保数据库账号拥有读取表结构的权限，否则方法二中的autoload_with会失败
• 两种方法都支持多列主键，你的单列主键场景完全适用
• 无需在多个插入函数中硬编码主键列名，实现了自动适配不同表的需求

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