本文目录导读：
我来详细介绍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是目前最推荐的选择，功能全面且维护活跃。
