Psycopg2导入CSV时，如何忽略含无效外键的行并继续事务？

要实现记录无效行、跳过插入且继续执行的需求，你可以先单独校验外键的有效性，再决定是否执行插入操作，同时收集无效行信息。以下是具体实现方案：

核心思路 #

1. 对每一行数据，先查询对应的外键ID是否存在
2. 若外键ID不存在（返回null），记录该行信息并跳过插入
3. 若外键ID有效，执行插入操作
4. 全程使用参数化查询避免SQL注入风险，同时捕获可能的异常保证流程不中断

代码实现 #

import pandas as pd

# 初始化列表存储无效行信息
invalid_rows = []

# 遍历CSV数据行
for index, row in df.iterrows():
    try:
        # 先查询对应的train_id是否存在
        check_query = f"SELECT id FROM {train_table} WHERE train_no = %s LIMIT 1"
        cursor.execute(check_query, (row[1],))
        train_id_result = cursor.fetchone()
        
        # 判断外键是否有效
        if not train_id_result:
            invalid_rows.append({
                "row_index": index,
                "train_no": row[1],
                "raw_data": row.to_dict()
            })
            continue  # 跳过当前行插入
        
        # 外键有效，执行插入操作（使用参数化查询）
        insert_query = f"""
            INSERT INTO {table_name} (travel_date, train_id, delay)
            VALUES (%s, %s, %s)
        """
        cursor.execute(insert_query, (row[0], train_id_result[0], row[2]))
    
    except Exception as e:
        # 捕获其他异常（如数据格式错误等），记录异常信息
        invalid_rows.append({
            "row_index": index,
            "train_no": row[1],
            "raw_data": row.to_dict(),
            "error_msg": str(e)
        })
        continue

# 提交事务（如果你的连接未开启自动提交）
# connection.commit()

# 保存无效行到文件，方便后续排查
if invalid_rows:
    invalid_df = pd.DataFrame(invalid_rows)
    invalid_df.to_csv("invalid_import_rows.csv", index=False)
    print(f"处理完成，共跳过 {len(invalid_rows)} 条无效行，详情请查看 invalid_import_rows.csv")

关键说明 #

• 避免SQL注入：替换原代码的f-string直接拼接参数方式，改用%s作为占位符的参数化查询，这是数据库操作的最佳实践。
• 事务连续性：只要不主动触发rollback()，即使某行处理失败，后续行的操作仍会继续执行；最后统一提交事务即可保证有效数据的入库。
• 无效行记录：不仅记录行索引和原始数据，还可捕获异常信息，方便后续定位问题（比如是外键无效还是其他数据错误）。

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