本文目录导读：
- Django ORM (makemigrations + migrate)
- SQLAlchemy (Alembic)
- SQLModel (SQLAlchemy + Pydantic)
- Peewee (pwiz + migrate)
- Tortoise ORM (Aerich)
- 自定义迁移生成器
- 最佳实践建议
Python ORM迁移自动生成主要通过数据库迁移工具来实现，不同的ORM框架有不同的迁移方案,下面介绍几种主流方案：
Django ORM (makemigrations + migrate)
Django内置了完善的迁移系统：
# models.py
from django.db import models
class User(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField(unique=True)
created_at = models.DateTimeField(auto_now_add=True)
# 自动生成迁移文件
python manage.py makemigrations
# 应用迁移
python manage.py migrate
# 查看SQL语句
python manage.py sqlmigrate app_name 0001
SQLAlchemy (Alembic)
Alembic是SQLAlchemy的官方迁移工具：
安装和初始化
pip install alembic alembic init alembic
配置 env.py
# alembic/env.py
from alembic import context
from sqlalchemy import engine_from_config
from your_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()
自动生成迁移
# models.py
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(100))
email = Column(String(100), unique=True)
# 自动生成迁移
alembic revision --autogenerate -m "create_users"
# 应用迁移
alembic upgrade head
# 查看历史
alembic history
SQLModel (SQLAlchemy + Pydantic)
SQLModel也基于Alembic：
# models.py
from sqlmodel import SQLModel, Field
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(max_length=100)
email: str = Field(unique=True)
# 初始化Alembic
alembic init alembic
# 配置后自动生成迁移
alembic revision --autogenerate -m "init"
alembic upgrade head
Peewee (pwiz + migrate)
Peewee使用内置的迁移功能：
from peewee import *
from playhouse.migrate import *
# 数据库连接
db = SqliteDatabase('test.db')
migrator = SqliteMigrator(db)
# 自动生成模型表
from peewee import Model, CharField, DateTimeField
class User(Model):
name = CharField(max_length=100)
email = CharField(unique=True)
class Meta:
database = db
# 创建表（相当于迁移）
User.create_table()
# 手动迁移
migrate(
migrator.add_column('user', 'age', IntegerField(default=0)),
migrator.drop_column('user', 'temp_field')
)
Tortoise ORM (Aerich)
适用于异步框架：
# models.py
from tortoise import fields, models
class User(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=100)
email = fields.CharField(max_length=100, unique=True)
# 初始化Aerich
aerich init -t config.TORTOISE_ORM
aerich init-db
# 自动生成迁移
aerich migrate --name add_user
# 应用迁移
aerich upgrade
# 回滚
aerich downgrade
自定义迁移生成器
如果需要更灵活的控制,可以自己实现：
# migration_generator.py
import inspect
import os
from datetime import datetime
class MigrationGenerator:
def __init__(self, model_class, migration_dir="migrations"):
self.model = model_class
self.migration_dir = migration_dir
def generate(self):
"""自动生成迁移文件"""
table_name = self.model.__tablename__
columns = self._extract_columns()
migration_code = self._create_migration_code(table_name, columns)
# 写入文件
filename = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_create_{table_name}.py"
path = os.path.join(self.migration_dir, filename)
with open(path, 'w') as f:
f.write(migration_code)
return path
def _extract_columns(self):
"""提取模型字段信息"""
columns = {}
for name, column in self.model.__table__.columns.items():
columns[name] = {
'type': str(column.type),
'nullable': column.nullable,
'primary_key': column.primary_key,
'unique': column.unique,
'default': column.default
}
return columns
def _create_migration_code(self, table_name, columns):
"""生成迁移代码"""
return f'''
from sqlalchemy import create_engine, MetaData, Table, Column, {', '.join(set(col['type'].split('(')[0] for col in columns.values())) if columns else 'String'}
def upgrade(engine):
metadata = MetaData()
{table_name} = Table(
'{table_name}', metadata,
{self._generate_columns_code(columns)}
)
metadata.create_all(engine)
def downgrade(engine):
{table_name}.drop(engine)
'''
def _generate_columns_code(self, columns):
"""生成列定义代码"""
lines = []
for name, col in columns.items():
params = [f"Column('{name}', {col['type']}"]
if not col['nullable']:
params.append('nullable=False')
if col['primary_key']:
params.append('primary_key=True')
if col['unique']:
params.append('unique=True')
params.append(')')
lines.append(' ' + ', '.join(params))
return ',\n'.join(lines)
# 使用示例
generator = MigrationGenerator(User)
migration_file = generator.generate()
最佳实践建议
版本控制
# alembic.ini 配置
[alembic]
script_location = alembic
sqlalchemy.url = sqlite:///app.db
# 自动检测模型变化
def include_object(obj, name, type_, reflected, compare_to):
if type_ == "table":
return name not in ["excluded_table"]
return True
迁移检查
# 检查是否有未应用的迁移 alembic check # 生成空迁移（手动编写） alembic revision -m "custom_migration" # 查看当前版本 alembic current # 查看迁移历史 alembic history --verbose
CI/CD 集成
# .github/workflows/migrations.yml
name: Database Migrations
on: [push]
jobs:
migrate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Run migrations
run: |
alembic upgrade head
alembic check # 确保没有未生成的迁移
选择合适的迁移工具取决于你的ORM框架和项目需求：
- Django：使用内置makemigrations
- SQLAlchemy：推荐Alembic
- 小型项目：Peewee的简单迁移
- 异步框架：Tortoise ORM + Aerich
迁移自动生成的核心是比较当前模型状态与数据库实际状态的差异，然后生成相应的ALTER TABLE语句。
