Python实现PostgreSQL数据库迁移遇错误及数据缺失问题求助

一、初始错误：字符串格式化失败的原因与修复 #

这个错误通常是SQL语句占位符与传递的参数数量不匹配，或是误用了psycopg2不支持的占位符格式（比如用?代替了psycopg2标准的%s）。只要确保占位符数量和参数数量一致，且使用%s作为参数占位符，就能解决这个问题。

二、数据缺失问题的完整修复脚本 #

目标库与源库表结构差异较大，需要做数据转换和关联处理，以下是可直接运行的迁移脚本：

首先确保安装依赖：

pip install psycopg2-binary

迁移脚本：

import psycopg2
from psycopg2 import OperationalError, Error

def create_connection(db_name, db_user, db_password, db_host, db_port):
    connection = None
    try:
        connection = psycopg2.connect(
            database=db_name,
            user=db_user,
            password=db_password,
            host=db_host,
            port=db_port,
        )
        print(f"Successfully connected to {db_name}")
    except OperationalError as e:
        print(f"Error connecting to {db_name}: {e}")
    return connection

# 配置数据库连接参数（替换为你的实际信息）
source_config = {
    "db_name": "source_db",
    "db_user": "your_user",
    "db_password": "your_password",
    "db_host": "localhost",
    "db_port": "5432"
}

dest_config = {
    "db_name": "destination_db",
    "db_user": "your_user",
    "db_password": "your_password",
    "db_host": "localhost",
    "db_port": "5432"
}

def migrate_data():
    source_conn = create_connection(**source_config)
    dest_conn = create_connection(**dest_config)
    
    if not source_conn or not dest_conn:
        return
    
    source_cursor = source_conn.cursor()
    dest_cursor = dest_conn.cursor()
    
    try:
        # 1. 迁移地址数据：源库Customers的address映射到目标库Addresses
        source_cursor.execute("SELECT customer_id, address FROM Customers")
        customers_addresses = source_cursor.fetchall()
        
        address_map = {}  # 记录源customer_id到目标address_id的对应关系
        for customer_id, address in customers_addresses:
            dest_cursor.execute(
                "INSERT INTO Addresses (street, city, state, country) VALUES (%s, %s, %s, %s) RETURNING address_id",
                (address, 'Unknown', 'Unknown', 'Unknown')
            )
            address_id = dest_cursor.fetchone()[0]
            address_map[customer_id] = address_id
        
        # 2. 迁移客户数据：拆分name为first_name和last_name
        source_cursor.execute("SELECT customer_id, name, email FROM Customers")
        customers = source_cursor.fetchall()
        
        for customer_id, name, email in customers:
            name_parts = name.split(maxsplit=1)
            first_name = name_parts[0] if len(name_parts) >= 1 else ''
            last_name = name_parts[1] if len(name_parts) >= 2 else ''
            
            dest_cursor.execute(
                """INSERT INTO Customers (customer_id, first_name, last_name, email, phone, address_id)
                   VALUES (%s, %s, %s, %s, %s, %s)""",
                (customer_id, first_name, last_name, email, 'Unknown', address_map[customer_id])
            )
        
        # 3. 迁移订单数据：补充缺失的order_date和is_delivered字段
        source_cursor.execute("SELECT order_id, customer_id, product, quantity, price FROM Orders")
        orders = source_cursor.fetchall()
        
        for order_id, customer_id, product, quantity, price in orders:
            dest_cursor.execute(
                """INSERT INTO Orders (order_id, customer_id, product, quantity, price, order_date, is_delivered)
                   VALUES (%s, %s, %s, %s, %s, CURRENT_DATE, %s)""",
                (order_id, customer_id, product, quantity, price, False)
            )
        
        dest_conn.commit()
        print("Data migration completed successfully!")
    
    except Error as e:
        dest_conn.rollback()
        print(f"Error occurred during data migration: {e}")
    finally:
        source_cursor.close()
        dest_cursor.close()
        source_conn.close()
        dest_conn.close()

if __name__ == "__main__":
    migrate_data()

三、关键修复点说明 #

• 字符串格式化错误修复：使用psycopg2标准的%s占位符，确保每个占位符对应一个参数，参数以元组形式传递，彻底避免参数不匹配问题。
• 客户表字段补全：通过split(maxsplit=1)拆分源库name字段为first_name和last_name，同时处理了只有名没有姓的边界情况。
• 地址关联处理：先迁移地址数据并建立映射关系，保证客户表的address_id字段能正确关联到地址表。
• 订单表缺失字段处理：为order_date设置当前日期，is_delivered默认设为False，你可以根据业务需求调整这些默认值。

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