Python实现PostgreSQL表插入时的间接引用处理方法

问题背景 #

现有两个CSV文件：country.csv（存储国家编码、名称及ID）和users.csv（存储用户信息及关联的Country-ID），已创建PostgreSQL的Country和Users表，其中Users的ISO_3166字段是关联Country表主键的外键。当前Python脚本可正常插入Country表数据，但插入Users表时，无法将users.csv中的Country-ID转换为对应的ISO_3166值，导致外键约束报错。

CSV与表结构参考 #

country.csv内容

Country Code,Country Name,Country-ID
US,United States,0
DE,Germany,1
AU,Australia,2
CZ,Czechia,3
CA,Canada,4
AR,Argentina,5
BR,Brazil,6
PT,Portugal,7
GB,United Kingdom,8
IT,Italy,9
GG,Guernsey,10
RO,Romania,11

users.csv内容

User-ID,Age,username,Country-ID
1,,madMeerkat6#yHazv,0
2,18.0,innocentUnicorn8#eCMNj,1
3,,jubilantStork8#YgoL-,0
4,17.0,hushedOatmeal4#y5QVW,0
5,,thrilledRhino7#3PYN3,0
6,61.0,insecureCaviar4#xosWW,0
7,,artisticGarlic3#Sla7S,2
8,,dearMandrill9#c1J0m,1
9,,cynicalDinosaur3#0wSxC,0
10,26.0,gloomyCake2#eRcdC,0
11,14.0,sincereCockatoo6#eDuI_,0

PostgreSQL表创建语句

CREATE TABLE Country (
    ISO_3166 CHAR(2) PRIMARY KEY,
    CountryName VARCHAR(256),
    CID varchar(16)
);
CREATE TABLE Users (
    UID INT PRIMARY KEY,
    Username VARCHAR(256),
    DoB DATE,
    Age INT,
    ISO_3166 CHAR(2) REFERENCES Country (ISO_3166)
);

解决方案 #

方案一：Python本地构建映射字典 #

读取country.csv时，同步构建Country-ID到ISO_3166的映射字典，处理users.csv时直接通过字典获取对应外键值。

修改后的Python脚本：

import csv
import psycopg2

def csv_to_dictionary(csv_name, delimiter):
    input_file = csv.DictReader(open(csv_name, 'r', encoding='utf-8'), delimiter=delimiter)
    return input_file

sql_con = psycopg2.connect(host='localhost', port='5432', database="XYZ", user='postgres', password='XYZ')
cursor = sql_con.cursor()

# 读取Country数据并插入，同时构建Country-ID到ISO_3166的映射
country_id_to_iso = {}
country_dictionary = csv_to_dictionary("country.csv", ',')
for row in country_dictionary:
    iso_code = row["Country Code"]
    country_id = row["Country-ID"]
    country_id_to_iso[country_id] = iso_code
    cursor.execute(""" INSERT INTO country (iso_3166, countryname, cid) VALUES (%s, %s, %s) """, 
                  (iso_code, row["Country Name"], country_id))

# 处理Users数据插入，通过映射字典获取正确的ISO_3166值
user_dictionary = csv_to_dictionary("users.csv", ',')  # 修正原脚本文件名错误：user.csv → users.csv
for row in user_dictionary:
    uid = int(row["User-ID"])
    username = row["username"]
    age = int(float(row["Age"])) if row["Age"] else None
    country_id = row["Country-ID"]
    iso_code = country_id_to_iso.get(country_id)

    # 动态构建插入语句，简化多分支判断
    fields = ["uid", "username"]
    values = [uid, username]
    if age is not None:
        fields.append("age")
        values.append(age)
    if iso_code is not None:
        fields.append("iso_3166")
        values.append(iso_code)
    
    placeholders = ", ".join(["%s"] * len(values))
    insert_sql = f""" INSERT INTO users ({", ".join(fields)}) VALUES ({placeholders}) """
    cursor.execute(insert_sql, values)

sql_con.commit()
cursor.close()
sql_con.close()

方案二：通过SQL临时表+关联查询插入 #

若数据量较大，可先将users.csv导入PostgreSQL临时表，再通过关联Country表直接插入Users表，无需在Python中处理映射。

步骤如下：

1. 创建临时表存储用户CSV数据：

CREATE TEMP TABLE temp_users (
    User_ID INT,
    Age NUMERIC,
    Username VARCHAR(256),
    Country_ID VARCHAR(16)
);

2. 使用COPY命令导入users.csv（替换为实际文件路径）：

COPY temp_users FROM '/path/to/users.csv' WITH (FORMAT CSV, HEADER);

3. 关联Country表插入数据到Users：

INSERT INTO Users (UID, Username, Age, ISO_3166)
SELECT 
    tu.User_ID,
    tu.Username,
    CASE WHEN tu.Age IS NOT NULL THEN tu.Age::INT ELSE NULL END,
    c.ISO_3166
FROM temp_users tu
JOIN Country c ON tu.Country_ID = c.CID;

4. 清理临时表（可选）：

DROP TABLE temp_users;

这种方法适合大数据量场景，利用数据库的关联查询能力完成转换，效率更高。

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