Python脚本如何操作数据库迁移

本文目录导读：

1. 主流数据库迁移工具
2. 自定义迁移脚本
3. 高级迁移管理
4. 最佳实践
5. 实战示例：完整的迁移流程
6. 推荐选择

我来详细介绍Python操作数据库迁移的常用方法和工具。

主流数据库迁移工具

Alembic (最常用)

Alembic是SQLAlchemy的数据库迁移工具,是Python生态中最流行的选择。

# 安装
pip install alembic
# 初始化
alembic init alembic

基本使用：

# alembic/env.py 配置
from alembic import context
from sqlalchemy import engine_from_config
from myapp.models import Base  # 你的模型
target_metadata = Base.metadata
def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(url=url, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()
# 创建迁移文件
# alembic revision --autogenerate -m "add user table"
# 执行迁移
# alembic upgrade head
# 回滚
# alembic downgrade -1

Django Migrations (Django项目)

Django自带强大的迁移系统：

# 创建迁移
python manage.py makemigrations
# 执行迁移
python manage.py migrate
# 查看迁移状态
python manage.py showmigrations
# 回滚迁移
python manage.py migrate app_name 0001

Flyway (Java生态但支持Python)

# 使用flyway命令
flyway migrate
flyway undo
flyway info

自定义迁移脚本

使用SQLAlchemy直接操作

from sqlalchemy import create_engine, text
engine = create_engine('postgresql://user:pass@localhost/dbname')
# 执行原始SQL迁移
def run_migration():
    with engine.connect() as conn:
        conn.execute(text("""
            CREATE TABLE IF NOT EXISTS users (
                id SERIAL PRIMARY KEY,
                username VARCHAR(50) NOT NULL,
                email VARCHAR(100) NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """))
        # 添加新列
        try:
            conn.execute(text("ALTER TABLE users ADD COLUMN age INTEGER"))
        except:
            pass  # 处理已存在情况
        conn.commit()

使用pymysql/psycopg2

import pymysql
def migrate_database():
    connection = pymysql.connect(
        host='localhost',
        user='root',
        password='password',
        database='mydb'
    )
    try:
        with connection.cursor() as cursor:
            # 版本控制表
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS schema_migrations (
                    version VARCHAR(50) PRIMARY KEY,
                    applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
            # 检查是否已执行
            cursor.execute("SELECT version FROM schema_migrations WHERE version='002_add_age'")
            if not cursor.fetchone():
                # 执行迁移
                cursor.execute("ALTER TABLE users ADD COLUMN age INT DEFAULT 0")
                cursor.execute("INSERT INTO schema_migrations (version) VALUES ('002_add_age')")
                connection.commit()
    finally:
        connection.close()

高级迁移管理

Yoyo Migrations

轻量级的迁移工具：

# 安装
pip install yoyo-migrations
# 创建迁移文件 migrations/001_create_users.py
from yoyo import step
step("""
    CREATE TABLE users (
        id INTEGER PRIMARY KEY,
        username VARCHAR(50),
        email VARCHAR(100)
    )
""", """
    DROP TABLE IF EXISTS users
""")  # 回滚步骤
# 执行迁移
from yoyo import read_migrations, get_backend
backend = get_backend('postgres://user:pass@localhost/db')
migrations = read_migrations('./migrations')
backend.apply_migrations(backend.to_apply(migrations))

SQLAlchemy-Migrate

较老但稳定的工具：

from migrate import *
from sqlalchemy import create_engine
engine = create_engine('sqlite:///mydb.db')
# 创建迁移
from migrate.versioning import api
api.create(engine, 'migrations')
# 编写迁移脚本
def upgrade(migrate_engine):
    meta = MetaData(bind=migrate_engine)
    users = Table('users', meta,
        Column('id', Integer, primary_key=True),
        Column('name', String(50)),
    )
    users.create()
def downgrade(migrate_engine):
    meta = MetaData(bind=migrate_engine)
    users = Table('users', meta)
    users.drop()

最佳实践

版本控制迁移文件

# migrations/001_initial.py
from alembic import op
import sqlalchemy as sa
revision = '001'
down_revision = None
def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.String(50), nullable=False),
        sa.Column('email', sa.String(100), unique=True),
    )
def downgrade():
    op.drop_table('users')

自动化迁移检查

def check_pending_migrations():
    """检查待执行的迁移"""
    from alembic.config import Config
    from alembic.script import ScriptDirectory
    from alembic.runtime.migration import MigrationContext
    config = Config("alembic.ini")
    script = ScriptDirectory.from_config(config)
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix='sqlalchemy.'
    )
    with connectable.connect() as connection:
        context = MigrationContext.configure(connection)
        current_rev = context.get_current_revision()
        print(f"Current revision: {current_rev}")
        # 获取所有未执行的迁移
        for rev in script.walk_revisions():
            if rev.is_head and rev.revision != current_rev:
                print(f"Pending migration: {rev.revision}: {rev.doc}")

实战示例：完整的迁移流程

# migrate.py
import os
import sys
from alembic.config import Config
from alembic import command
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DatabaseMigrator:
    def __init__(self, alembic_cfg_path='alembic.ini'):
        self.alembic_cfg = Config(alembic_cfg_path)
    def create_migration(self, message):
        """创建新的迁移文件"""
        command.revision(self.alembic_cfg, 
                        autogenerate=True, 
                        message=message)
        logger.info(f"Created migration: {message}")
    def upgrade(self, revision='head'):
        """执行迁移"""
        command.upgrade(self.alembic_cfg, revision)
        logger.info(f"Migrated to: {revision}")
    def downgrade(self, revision='-1'):
        """回滚迁移"""
        command.downgrade(self.alembic_cfg, revision)
        logger.info(f"Downgraded to: {revision}")
    def current(self):
        """查看当前版本"""
        command.current(self.alembic_cfg)
    def history(self):
        """查看迁移历史"""
        command.history(self.alembic_cfg)
# 使用示例
if __name__ == '__main__':
    migrator = DatabaseMigrator()
    # 创建新迁移
    migrator.create_migration("add user table")
    # 执行到最新版本
    migrator.upgrade()
    # 回滚一步
    migrator.downgrade('-1')
    # 查看状态
    migrator.current()

推荐选择

• 新项目: Alembic (SQLAlchemy) 或 Django Migrations
• 简单项目: Yoyo或自定义脚本
• 需要事务支持: Alembic
• 非Python项目: Flyway

选择迁移工具时,考虑项目规模、团队熟悉度和数据库复杂度，Alembic是目前最推荐的选择，功能全面且维护活跃。