PythonORM迁移自动生成怎么实现

wen python案例 2026-07-06 176

本文目录导读：

PythonORM迁移自动生成怎么实现

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框架和项目需求：