使用Python pyodbc批量迁移SQL Server数据并仅删已迁移行的问题

原代码的核心问题 #

1. 执行语句错误：删除操作时误执行了插入的query而非delete_query，直接导致删除逻辑完全错误
2. 无批次限制：没有指定每次迁移的行数（如TOP 1000），会一次性迁移所有同日期数据，不符合分批需求
3. 匹配条件不精准：仅用日期作为匹配条件，若迁移过程中原表有同日期新数据写入，会误删未迁移的行
4. 异常处理过于宽泛：直接except:捕获所有异常，无法定位具体问题（如主键冲突、连接异常等）

精准匹配的实现方案 #

要确保只删除已成功迁移的行，最可靠的方式是通过**唯一标识（主键/唯一键）**关联迁移前后的数据，结合批次限制实现循环处理。以下是两种可行方案：

方案1：使用OUTPUT子句捕获已迁移行的主键

利用SQL Server的OUTPUT子句，在插入新表时直接返回已迁移行的主键，再用这些主键删除原表对应数据，完全保证迁移和删除的是同一批数据。

import pyodbc  # 假设使用pyodbc连接SQL Server

# 初始化连接和游标（示例）
conn = pyodbc.connect('DRIVER={SQL Server};SERVER=your_server;DATABASE=your_db;UID=user;PWD=pwd')
cursor = conn.cursor()

batch_size = 1000
table_name1 = "new_table"
table_name2 = "old_table"
database = "source_db"
column_names = "col1, col2, id"  # 包含主键字段id
primary_key = "id"  # 假设主键是id

while True:
    try:
        # 1. 插入TOP 1000数据，同时输出已插入的主键
        insert_query = f"""
            INSERT INTO [{table_name1}] ({column_names})
            OUTPUT inserted.{primary_key}
            SELECT TOP ({batch_size}) {column_names}
            FROM {database}.{table_name2}
            -- 可选：如果需要按日期过滤，加上WHERE条件
            -- WHERE date_column = 'target_date'
        """
        cursor.execute(insert_query)
        # 获取已插入的主键列表
        migrated_ids = [row[0] for row in cursor.fetchall()]
        conn.commit()

        # 如果没有数据插入，退出循环
        if not migrated_ids:
            break

        # 2. 删除原表中对应主键的行
        # 用参数化查询避免SQL注入，处理批量ID
        placeholders = ','.join(['?' for _ in migrated_ids])
        delete_query = f"""
            DELETE FROM {database}.{table_name2}
            WHERE {primary_key} IN ({placeholders})
        """
        cursor.execute(delete_query, migrated_ids)
        conn.commit()

        print(f"成功迁移并删除 {len(migrated_ids)} 条数据")

    except pyodbc.Error as e:
        print(f"执行出错：{e}")
        conn.rollback()
        break

# 关闭连接
cursor.close()
conn.close()

方案2：先查询批次主键，再迁移删除

如果不使用OUTPUT，可以先查询原表TOP 1000的主键，根据主键迁移数据，再删除对应行，同样能保证精准匹配。

while True:
    try:
        # 1. 查询TOP 1000待迁移的主键
        select_ids_query = f"""
            SELECT TOP ({batch_size}) {primary_key}
            FROM {database}.{table_name2}
            -- 可选：WHERE date_column = 'target_date'
        """
        cursor.execute(select_ids_query)
        target_ids = [row[0] for row in cursor.fetchall()]

        if not target_ids:
            break

        # 2. 根据主键插入数据到新表
        placeholders = ','.join(['?' for _ in target_ids])
        insert_query = f"""
            INSERT INTO [{table_name1}] ({column_names})
            SELECT {column_names}
            FROM {database}.{table_name2}
            WHERE {primary_key} IN ({placeholders})
        """
        cursor.execute(insert_query, target_ids)
        conn.commit()

        # 3. 删除原表对应主键的行
        delete_query = f"""
            DELETE FROM {database}.{table_name2}
            WHERE {primary_key} IN ({placeholders})
        """
        cursor.execute(delete_query, target_ids)
        conn.commit()

        print(f"成功迁移并删除 {len(target_ids)} 条数据")

    except pyodbc.Error as e:
        print(f"执行出错：{e}")
        conn.rollback()
        break

关键注意事项 #

• 避免SQL注入：始终使用参数化查询（?占位符），不要用f-string直接拼接用户输入或变量到SQL语句中
• 事务控制：每次迁移和删除操作放在同一事务中，确保要么都成功要么都回滚，避免数据不一致
• 主键/唯一键依赖：必须依赖表的主键或唯一键来关联数据，仅用日期等非唯一条件无法保证精准匹配
• 异常处理：捕获具体的数据库异常（如pyodbc.Error），方便定位问题

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