如何将MongoDB的ObjectId插入PostgreSQL作为主键？

问题原因 #

你生成的SQL语句中，_id的ObjectId值（如6375364d1ad809ab6108a544）未被单引号包裹。PostgreSQL会将其识别为数字字面量，但值中包含字母，因此触发语法错误。而你的PostgreSQL_id字段类型为VARCHAR，字符串类型的值必须用单引号括起来。

修复方案 #

方案1：临时修改字符串拼接（不推荐，存在SQL注入风险） #

在拼接SQL时，给非NULL的字符串值添加单引号，并转义值中的单引号：

def insert_psql(db, table_name: str, fields: dict, data: list):
    new_fields = {}
    for field in fields:
        new_fields[field.replace('.', '_')] = fields[field]
    insert_query = f'INSERT INTO {table_name} ('
    insert_query += ', '.join(new_fields.keys()) + ')'
    insert_query += 'VALUES '
    for i, row in enumerate(data):
        insert_query += '('
        for j, field in enumerate(fields):
            if not row.get(field):
                insert_query += 'NULL'
            else:
                value = str(row.get(field))
                # 转义单引号，避免SQL语法错误
                escaped_value = value.replace("'", "''")
                insert_query += f"'{escaped_value}'"
            if not j == len(fields) - 1:
                insert_query += ', '
        insert_query += ')'
        if not i == len(data) - 1:
            insert_query += ', '
    try:
        db.execute(insert_query)
        db.commit()
    except Exception as e:
        print(e)

方案2：使用参数化查询（推荐，安全高效） #

psycopg2支持参数化查询，能自动处理字符串引号和转义，还能避免SQL注入，批量插入效率更高：

from bson.objectid import ObjectId

def insert_psql(db, table_name: str, fields: dict, data: list):
    # 处理字段名中的点号
    new_fields = {field.replace('.', '_'): fields[field] for field in fields}
    field_names = ', '.join(new_fields.keys())
    # 生成参数占位符
    placeholders = ', '.join(['%s'] * len(new_fields))
    # 添加冲突处理，避免重复行（_id作为主键）
    insert_query = f'INSERT INTO {table_name} ({field_names}) VALUES ({placeholders}) ON CONFLICT (_id) DO NOTHING'
    
    # 准备批量插入的参数列表
    params_list = []
    for row in data:
        row_params = []
        for field in fields:
            val = row.get(field)
            # 将MongoDB ObjectId转为字符串
            if isinstance(val, ObjectId):
                row_params.append(str(val))
            else:
                row_params.append(val if val is not None else None)
        params_list.append(tuple(row_params))
    
    try:
        # 执行批量插入
        db.executemany(insert_query, params_list)
        db.commit()
    except Exception as e:
        print(e)
        db.rollback()

补充说明 #

• 若要将_id作为主键，需确保PostgreSQL表中_id字段已设置为PRIMARY KEY。
• ON CONFLICT (_id) DO NOTHING表示当_id重复时跳过该行插入；若需要更新现有行，可改为ON CONFLICT (_id) DO UPDATE SET 字段1 = EXCLUDED.字段1, 字段2 = EXCLUDED.字段2。

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