Python SQLAlchemy如何对比表/行并识别数据差异（Deltas）？

Great question! 我之前处理过类似的数据同步场景，下面分享几种用SQLAlchemy识别数据差异并执行INSERT/UPDATE的实用方法：

你提到的Python集合思路完全可以和SQLAlchemy结合，逻辑直观，和你熟悉的集合操作一致。核心是把两张表的数据拉到内存，转换成可哈希的集合（比如元组），再通过集合运算找出差异：

from sqlalchemy import create_engine, Table, MetaData

# 初始化数据库连接
engine = create_engine("your-database-url")
metadata = MetaData()

# 加载两张表结构（假设表结构一致，且第一列是主键）
table1 = Table("table1", metadata, autoload_with=engine)
table2 = Table("table2", metadata, autoload_with=engine)

# 拉取数据并转换为集合
with engine.connect() as conn:
    table1_rows = {(row[0], row[1]) for row in conn.execute(table1.select()).all()}
    table2_rows = {(row[0], row[1]) for row in conn.execute(table2.select()).all()}

# 区分新增和更新行
# 新增行：table2存在但table1没有的主键
existing_pks = {row[0] for row in table1_rows}
insert_rows = [row for row in table2_rows if row[0] not in existing_pks]
# 更新行：主键存在但字段值不同的行
updated_rows = [row for row in table2_rows if row[0] in existing_pks and row not in table1_rows]

# 执行同步操作
with engine.begin() as conn:
    # 批量更新
    for pk_val, new_val in updated_rows:
        conn.execute(
            table1.update()
            .where(table1.c.id == pk_val)
            .values(second_col=new_val)
        )
    # 批量插入
    conn.execute(table1.insert(), [{"id": pk, "second_col": val} for pk, val in insert_rows])

这种方法的优势是逻辑简单易懂，但只适合数据量不大的场景——毕竟要把数据全量拉到内存里。

如果不想自己写集合逻辑，可以用datacompy这个专门对比数据的工具，它能直接和SQLAlchemy集成，自动识别新增、更新和差异行：

import datacompy
from sqlalchemy import create_engine

engine = create_engine("your-database-url")

# 直接对比两张表，指定主键作为匹配依据
comparison = datacompy.compare(
    engine,
    "SELECT * FROM table1",
    "SELECT * FROM table2",
    join_columns="id"
)

# 提取差异结果
insert_rows = comparison.df2_unq_rows.to_dict("records")  # table2独有的行（新增）
updated_rows = comparison.mismatch_rows.to_dict("records")  # 主键匹配但字段不同的行（更新）

# 执行同步
with engine.begin() as conn:
    # 更新操作
    for row in updated_rows:
        conn.execute(
            table1.update()
            .where(table1.c.id == row["id"])
            .values(second_col=row["second_col"])
        )
    # 插入操作
    conn.execute(table1.insert(), insert_rows)

datacompy会帮你处理很多细节，比如字段类型匹配、空值判断，比自己写集合运算更健壮。

如果数据量很大，拉到内存不现实，那就用你一开始想到的外连接，直接在数据库里计算差异，再通过SQLAlchemy执行语句：

from sqlalchemy import create_engine, Table, MetaData, select, update, insert

engine = create_engine("your-database-url")
metadata = MetaData()
table1 = Table("table1", metadata, autoload_with=engine)
table2 = Table("table2", metadata, autoload_with=engine)

# 1. 生成更新语句：匹配主键但字段不同的行
update_stmt = (
    update(table1)
    .values(second_col=table2.c.second_col)
    .where(table1.c.id == table2.c.id)
    .where(table1.c.second_col != table2.c.second_col)
)

# 2. 生成插入语句：table2有但table1没有的行
insert_stmt = (
    insert(table1)
    .from_select(
        ["id", "second_col"],
        select(table2.c.id, table2.c.second_col)
        .where(~table2.c.id.in_(select(table1.c.id)))
    )
)

# 执行操作
with engine.begin() as conn:
    conn.execute(update_stmt)
    conn.execute(insert_stmt)

这种方法把计算压力交给数据库，不需要拉取全量数据，效率更高，适合大数据量的同步场景。

你提到的SQLAlchemy Core中的compare相关功能，其实主要是用来对比表结构（Schema）差异的，比如Table.compare()方法可以检查两个表的字段、约束是否一致，并不是用来对比数据内容的。所以如果要处理数据差异，还是上面几种方法更实用。

总结选择建议： #

• 小数据量：用集合运算+SQLAlchemy，逻辑直观
• 中等数据量：用datacompy，省心省力
• 大数据量：用数据库外连接，效率优先

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