Python操作PostgreSQL如何替换代码内的for循环实现等价业务逻辑

原有代码的核心性能瓶颈如下：

• 嵌套循环匹配已存记录：每次遍历全部数据库查询结果，时间复杂度为O(CSV行数 * 数据库已有记录数)，数据量超过万级后卡顿非常明显
• 单条执行SQL+逐次commit：每处理一行就和数据库做一次IO交互，且commit是阻塞持久化操作，占整体运行时间的90%以上
• 逐字段赋值逻辑冗余，无必要的重复计算多

以下是可完全替代原有逻辑的Python层面实现方案，效率可提升数十到数百倍：

方案1：字典映射替换内层循环 + 批量SQL执行（改动最小，无额外依赖） #

无需引入第三方库，仅修改原有逻辑结构即可：

1. 先将查询到的usa_records转为字典，用匹配字段（索引6）作为key，O(1)时间即可完成匹配，直接干掉内层循环
2. 先把所有待更新、待插入的数据暂存到列表，最后批量执行SQL，统一commit，减少IO次数

示例代码：

import csv
import psycopg2

# 1. 先将数据库已有记录转为字典，直接消除内层循环
usa_records = cursor.fetchall()
# 匹配键为索引6的字段，值为整条记录
usa_map = {record[6]: record for record in usa_records}

update_data = []
insert_data = []

# 2. 遍历CSV文件
for k in csv_files:
    with open(k, 'r', encoding='utf-8') as f:
        reader = csv.reader(f)
        next(reader)  # 跳过表头
        for row in reader:
            match_record = usa_map.get(row[6])
            if match_record:
                # 存在匹配记录，合并字段
                for i in range(18):
                    if match_record[i] is not None:
                        row[i] = match_record[i]
                # 按参数顺序组装更新数据，最后一个参数是id（row[17]）
                update_data.append(tuple(row[:17] + [row[17]]))
            else:
                # 无匹配，加入插入列表
                insert_data.append(tuple(row))

# 3. 批量执行更新
if update_data:
    sql_update_query = """Update usa set technology=%s, company_name=%s,contact_name=%s,first_name=%s,last_name=%s,title=%s,email=%s,person_linkedin_url=%s,web_address=%s,company_linkedin_url=%s,company_address=%s,city=%s,state=%s,company_phone=%s,employees=%s,industry=%s,country=%s where id = %s"""
    cursor.executemany(sql_update_query, update_data)
    print(f"{len(update_data)} 条记录更新完成")

# 4. 批量执行插入
if insert_data:
    sql_insert_query = "INSERT INTO usa VALUES (%s, %s, %s, %s,%s, %s, %s, %s,%s, %s, %s, %s,%s, %s, %s, %s,%s,%s)"
    cursor.executemany(sql_insert_query, insert_data)
    print(f"{len(insert_data)} 条记录插入完成")

# 统一提交，仅做一次commit
conn.commit()

方案2：Pandas + PostgreSQL原生UPSERT（开发效率最高，性能优异） #

如果可以引入第三方库，直接用pandas批量读取所有CSV，配合PG的INSERT ON CONFLICT语法，连提前查询数据库已有记录的步骤都可以省掉，完全不需要在Python侧做匹配逻辑：

1. 先给数据库中用于匹配的字段（即原代码中索引6对应的列，通常是email）添加唯一约束
2. 用pandas一次性读取所有CSV合并为一个DataFrame
3. 生成带冲突更新的批量插入语句，直接执行即可，数据库会自动处理"存在则更新、不存在则插入"的逻辑

示例代码：

import pandas as pd
from sqlalchemy import create_engine

# 连接数据库
engine = create_engine('postgresql://用户名:密码@主机:端口/库名')

# 批量读取所有CSV合并为一个DataFrame
df_list = []
for k in csv_files:
    df = pd.read_csv(k)
    df_list.append(df)
all_df = pd.concat(df_list, ignore_index=True)

# 生成UPSERT语句，注意替换括号里的匹配字段名、对应更新的字段名
upsert_sql = """
INSERT INTO usa (technology, company_name, contact_name, first_name, last_name, title, email, person_linkedin_url, web_address, company_linkedin_url, company_address, city, state, company_phone, employees, industry, country, id)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (email) DO UPDATE SET
technology = COALESCE(EXCLUDED.technology, usa.technology),
company_name = COALESCE(EXCLUDED.company_name, usa.company_name),
-- 剩余字段按上面的格式补全即可，COALESCE逻辑就是原代码的非空才覆盖的逻辑
country = COALESCE(EXCLUDED.country, usa.country)
"""

# 批量执行
with engine.connect() as conn:
    conn.execute(upsert_sql, all_df.values.tolist())
    conn.commit()

方案3：psycopg2的copy_from接口（性能最高，适合百万级以上大数据量） #

超大数据量场景下可以用PG的COPY命令，比executemany还要快数倍：

1. 先将所有CSV数据写入临时表
2. 用SQL直接从临时表批量UPSERT到正式表，全程几乎没有Python层面的循环开销

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