使用Python脚本从MongoDB迁移数据到Postgres无法匹配对应列咨询

你的现有代码将整条MongoDB文档序列化为JSON字符串后作为唯一插入值，因此只会写入表的第一列。要实现字段匹配写入对应列，需要将文档按键值拆分，匹配表的列名后用参数化查询插入，修正后的代码如下：

import json
import psycopg2
from pymongo import MongoClient
from decimal import Decimal
from bson.decimal128 import Decimal128
from bson.objectid import ObjectId
from datetime import datetime
import time

class JSONEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, ObjectId):
            return str(o)
        if isinstance(o, Decimal):
            return float(o)
        if isinstance(o, Decimal128):
            return float(str(o))
        if isinstance(o, datetime):
            return str(o)
        return json.JSONEncoder.default(self, o)

def convert_value(val):
    # 统一转换MongoDB特殊类型为Postgres可识别的格式
    if isinstance(val, ObjectId):
        return str(val)
    if isinstance(val, (Decimal, Decimal128)):
        return float(str(val))
    if isinstance(val, datetime):
        return str(val)
    return val

client = MongoClient("mongodb+srv://reporting:reporting123@miramira.ntara.mongodb.net")
db=client.Bookstore
col=db.Book
mdbcur=col.find()
with psycopg2.connect(host="localhost",
    database="Book",
    user="postgres", 
    password="root") as pgconn:
    pgcur=pgconn.cursor()
    for doc in mdbcur:
        # 转换所有字段值为Postgres兼容格式
        processed_doc = {k: convert_value(v) for k, v in doc.items()}
        # 动态获取列名和对应值
        columns = list(processed_doc.keys())
        values = list(processed_doc.values())
        # 生成参数化查询语句，占位符数量和字段数量匹配
        placeholders = ', '.join(['%s'] * len(columns))
        query_sql = f"INSERT INTO booktable ({', '.join(columns)}) VALUES ({placeholders})"
        pgcur.execute(query_sql, values)
        # 若需要逐行提交则打开下方注释
        # pgconn.commit()
        time.sleep(1)
    # 统一提交所有插入操作，效率更高
    pgconn.commit()
    pgcur.close()
    mdbcur.close()

注意事项 #

• 请先确保Postgres中booktable的列名和MongoDB返回的文档字段名完全一致，如果存在名称差异，需要自行新增字段映射逻辑做名称转换
• 采用参数化查询替代原有的字符串拼接SQL逻辑，避免了特殊字符导致的语法错误和SQL注入风险
• 代码默认采用批量提交逻辑，如果你需要逐行持久化数据，可以将pgconn.commit()移回循环内部
• 如果你需要所有字段都以字符串格式写入Postgres，可修改convert_value函数，所有返回值都用str()包裹即可

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