【Python】如何进行数据库表迁移：一种简单的方法

如何进行数据库表迁移：一种简单的方法

在系统的开发与维护过程中，数据库表迁移是不可避免的任务之一。无论是为了更换数据库引擎、结构优化，还是业务扩展，迁移数据需要谨慎处理。本文将介绍如何通过编写 Python 脚本，进行两个数据库之间的数据迁移。

1. 准备工作

首先，我们需要准备两个数据库——源数据库和目标数据库。以 PostgreSQL 为例，可以使用 SQLAlchemy 来处理数据库的连接和查询。使用 SQLAlchemy 可以方便地进行 ORM 映射，并简化数据迁移的操作。

安装依赖：

pip install sqlalchemy psycopg2

定义源数据库和目标数据库的模型（这里只展示迁移逻辑的简化版）：

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 创建ORM映射的基类
Base = declarative_base()

# 源数据库模型
class SourceTable(Base):
    __tablename__ = 'source_table'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    value = Column(Integer)

# 目标数据库模型
class TargetTable(Base):
    __tablename__ = 'target_table'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    value = Column(Integer)

# 创建数据库连接
source_engine = create_engine('postgresql://user:password@localhost/source_db')
target_engine = create_engine('postgresql://user:password@localhost/target_db')

# 创建会话
SourceSession = sessionmaker(bind=source_engine)
TargetSession = sessionmaker(bind=target_engine)

在这里，我们定义了两个数据库的模型类 SourceTable 和 TargetTable，它们结构相同，包含 id、name 和 value 三个字段。

2. 编写迁移脚本

接下来，我们编写数据迁移逻辑，将源数据库的数据读取出来，并插入到目标数据库中。

def migrate_data():
    source_session = SourceSession()
    target_session = TargetSession()

    try:
        # 查询源数据库中的所有数据
        data_to_migrate = source_session.query(SourceTable).all()

        for record in data_to_migrate:
            # 检查目标数据库中是否已有记录
            existing_record = target_session.query(TargetTable).filter_by(id=record.id).first()

            if existing_record:
                print(f"Record with id {record.id} already exists, skipping.")
                continue
            
            # 创建目标表的记录
            new_record = TargetTable(
                id=record.id,
                name=record.name,
                value=record.value
            )

            # 添加到目标数据库
            target_session.add(new_record)

        # 提交事务
        target_session.commit()
        print("Data migration completed successfully!")

    except Exception as e:
        target_session.rollback()
        print(f"Migration failed: {e}")
    
    finally:
        source_session.close()
        target_session.close()

if __name__ == "__main__":
    migrate_data()