如何将SQLAlchemy表中指定字段迁移至另一表（需去重）

看起来你已经习惯用字典初始化ORM实例的方式添加记录了，针对你要从SourceDB迁移name和place字段到DestDB，同时避免name+place重复的需求，有两种更简洁高效的方案，比逐条遍历转字典要好用得多：

方案一：ORM层方案（符合你的字典使用习惯） #

这个方案先把目标表中已有的name+place组合捞出来，再过滤源表中不存在的记录，最后批量转换为DestDB实例添加，逻辑直观，适合数据量不大的场景：

from sqlalchemy import tuple_

# 先获取目标表中已存在的(name, place)组合，用集合存起来做快速查重
existing_pairs = {(row.name, row.place) for row in session.query(DestDB.name, DestDB.place)}

# 查询源表中不在已有组合里的记录，只取需要的两个字段
source_records = session.query(SourceDB.name, SourceDB.place).filter(
    ~tuple_(SourceDB.name, SourceDB.place).in_(existing_pairs)
).all()

# 把查询结果（namedtuple）转成字典，再初始化DestDB实例，批量添加
dest_instances = [DestDB(**row._asdict()) for row in source_records]
session.add_all(dest_instances)
session.commit()

这里row._asdict()会自动把SQLAlchemy的查询结果（本质是命名元组）转换成你熟悉的字典格式，完美适配你习惯的**dict初始化方式。

方案二：核心层批量插入（高效适合大数据量） #

如果源表数据量很大，把所有记录拉到内存处理会很占资源，这时候用SQLAlchemy核心层的冲突处理语法，直接让数据库层面处理去重，效率高很多：

PostgreSQL版本 #

from sqlalchemy import insert

# 构建源表的查询语句，只取需要的字段，如果源表本身有重复，记得加.distinct()去重
source_select = session.query(SourceDB.name, SourceDB.place).distinct().statement

# 插入目标表，遇到联合索引冲突就跳过
insert_stmt = insert(DestDB).from_select(
    ['name', 'place'],
    source_select
).on_conflict_do_nothing(
    index_elements=['nameplace']  # 这里用你在DestDB里定义的联合索引名
)

# 执行语句并提交
session.execute(insert_stmt)
session.commit()

MySQL版本 #

MySQL没有on_conflict_do_nothing，可以用on_duplicate_key_update做个“空操作”来实现相同效果：

insert_stmt = insert(DestDB).from_select(
    ['name', 'place'],
    source_select
).on_duplicate_key_update(name=DestDB.name)  # 原地更新相同值，相当于不做任何操作

session.execute(insert_stmt)
session.commit()

额外提示 #

如果源表本身存在name+place重复的记录，记得在查询源表时加上.distinct()，避免把重复数据带到目标表。

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