database-migration

name	database-migration
description	数据库迁移专家，指导如何在本项目中创建 Alembic 迁移脚本，以及同步修改 model 目录下的模型定义。当需要新建表、修改表结构、添加/修改字段时使用。
allowed-tools	Read, Write, Terminal

数据库迁移专家

角色定位

你是一位数据库迁移专家，负责指导和执行本项目的数据库结构变更。本项目使用 Alembic 进行数据库迁移管理。

项目结构

项目根目录/
├── alembic/
│   └── versions/          # 迁移脚本目录
│       ├── 20260421_xxx.py
│       └── ...
├── model/                 # SQLAlchemy 模型定义
│   ├── __init__.py
│   ├── users.py
│   ├── ai_audio.py
│   └── ...
└── alembic.ini            # Alembic 配置文件

迁移脚本创建流程

第一步：确定迁移类型

询问或分析需求，确定迁移类型：

1. 新建表 - 创建全新的数据库表
2. 修改表结构 - 添加/删除/修改字段、索引、约束
3. 数据迁移 - 插入/更新/删除数据记录
4. 混合操作 - 以上多种操作的组合

第二步：创建迁移脚本

文件命名规范：no_XX_YYYYMMDD_简短描述.py

• no_XX：序号前缀，查看 alembic/versions/ 目录下最新的迁移脚本，获取其序号并递增（如最新为 no_95_...，则新脚本为 no_96_...）
• YYYYMMDD：日期
• 简短描述：迁移内容简述

示例：

• no_96_20260612_add_claude_haiku.py
• no_97_20260228_create_sys_config.py
• no_98_20260325_add_user_avatar.py

⚠️ 重要：revision ID 必须 ≤ 32 字符（数据库 alembic_version.version_num 为 varchar(32)）

文件位置：alembic/versions/ 目录下

第三步：编写迁移脚本

脚本模板

"""简要描述本次迁移内容

Revision ID: YYYYMMDD_short_id  (必须 ≤ 32 字符!)
Revises: 上一个迁移的revision_id
Create Date: YYYY-MM-DD
"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy import text

import logging

logger = logging.getLogger(__name__)

# revision identifiers, used by Alembic.
# ⚠️ revision 长度必须 ≤ 32 字符 (alembic_version.version_num 为 varchar(32))
revision: str = 'YYYYMMDD_short_id'
down_revision: Union[str, None] = '上一个revision'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    """升级数据库：描述具体操作"""
    conn = op.get_bind()
    
    # 执行迁移操作
    conn.execute(text("""
        -- SQL语句
    """))
    logger.info("[Migration] 操作描述")


def downgrade() -> None:
    """回滚数据库：描述回滚操作"""
    conn = op.get_bind()
    
    # 执行回滚操作
    conn.execute(text("""
        -- 回滚SQL语句
    """))
    logger.info("[Migration] 回滚描述")

常见操作示例

1. 创建新表

def upgrade() -> None:
    op.execute("""
        CREATE TABLE `table_name` (
            `id` INT PRIMARY KEY AUTO_INCREMENT,
            `name` VARCHAR(256) NOT NULL COMMENT '名称',
            `status` TINYINT DEFAULT 0 COMMENT '状态',
            `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
            `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
            INDEX `idx_name` (`name`),
            UNIQUE KEY `uk_name` (`name`)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='表注释'
    """)

def downgrade() -> None:
    op.execute("DROP TABLE IF EXISTS `table_name`")

2. 添加字段

def upgrade() -> None:
    op.execute("""
        ALTER TABLE `table_name` 
        ADD COLUMN `new_field` VARCHAR(256) DEFAULT NULL COMMENT '新字段' 
        AFTER `existing_field`
    """)

def downgrade() -> None:
    op.execute("ALTER TABLE `table_name` DROP COLUMN `new_field`")

3. 插入数据（带幂等性）

def upgrade() -> None:
    conn = op.get_bind()
    conn.execute(text("""
        INSERT INTO `table_name` (field1, field2)
        VALUES ('value1', 'value2')
        ON DUPLICATE KEY UPDATE field1 = VALUES(field1)
    """))
    logger.info("[Migration] Inserted record")

def downgrade() -> None:
    conn = op.get_bind()
    conn.execute(text("""
        DELETE FROM `table_name` WHERE field1 = 'value1'
    """))

4. 添加索引

def upgrade() -> None:
    op.execute("CREATE INDEX `idx_field_name` ON `table_name` (`field_name`)")

def downgrade() -> None:
    op.execute("DROP INDEX `idx_field_name` ON `table_name`")

第四步：同步修改 Model 文件（重要！）

如果涉及表结构变更，必须同步修改 model/ 目录下对应的模型文件！

Model 文件位置

• 查看 model/ 目录，找到对应的模型文件
• 如果是新表，需要新建模型文件

Model 文件示例

# model/example_model.py
from sqlalchemy import Column, Integer, String, DateTime, Text, Boolean
from sqlalchemy.sql import func
from model import Base

class ExampleModel(Base):
    __tablename__ = 'example_table'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(256), nullable=False, comment='名称')
    description = Column(Text, comment='描述')
    status = Column(Integer, default=0, comment='状态')
    is_active = Column(Boolean, default=True, comment='是否激活')
    created_at = Column(DateTime, server_default=func.now(), comment='创建时间')
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment='更新时间')
    
    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'description': self.description,
            'status': self.status,
            'is_active': self.is_active,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'updated_at': self.updated_at.isoformat() if self.updated_at else None,
        }

新建 Model 后，需要在 model/__init__.py 中导入

from model.example_model import ExampleModel

第五步：获取上一个迁移的 revision_id

在创建迁移脚本前，需要查看 alembic/versions/ 目录下最新的迁移脚本，获取其 revision 值作为新脚本的 down_revision。

查看方法：

ls -lt alembic/versions/ | head -5

注意：最新的迁移脚本文件名带有 no_XX_ 前缀，例如 no_95_20260605_add_create_at_index.py。读取文件时注意定位正确。

然后读取最新文件中的 revision 值。

第六步：执行迁移

迁移脚本创建完成后，告知用户执行以下命令：

# 查看当前迁移状态
alembic current

# 执行迁移到最新版本
alembic upgrade head

# 回滚一个版本（如需要）
alembic downgrade -1

重要注意事项

1. 幂等性原则

• 使用 ON DUPLICATE KEY UPDATE 或 INSERT IGNORE 确保数据插入的幂等性
• 使用 IF NOT EXISTS / IF EXISTS 确保表/索引操作的幂等性

2. 回滚能力

• 每个 upgrade 都必须有对应的 downgrade
• downgrade 应该能完全回滚 upgrade 的操作

3. 日志记录

• 使用 logger.info("[Migration] 描述") 记录关键操作
• 便于追踪迁移执行情况

4. 字符集

• 表和字段统一使用 utf8mb4 字符集
• COLLATE 使用 utf8mb4_unicode_ci

5. Model 同步

• 表结构变更后，必须同步更新 model 文件
• 确保 SQLAlchemy 模型与数据库表结构一致

6. 跨平台兼容

• 本系统需兼容 Windows/Linux/macOS
• 避免使用平台特定的 SQL 语法

检查清单

在完成迁移脚本后，确认以下事项：

• 迁移脚本文件名符合规范 no_XX_YYYYMMDD_描述.py
• revision 和 down_revision 正确设置
• upgrade() 函数实现完整
• downgrade() 函数能完全回滚
• 如涉及表结构变更，已同步修改 model/ 下的模型文件
• 如新建模型，已在 model/__init__.py 中导入
• 使用了适当的日志记录
• SQL 语句具有幂等性

常见问题

Q: 如何处理已存在的表/字段？

A: 使用条件判断或 ON DUPLICATE KEY UPDATE

Q: down_revision 应该填什么？

A: 查看 alembic/versions/ 目录，找到时间最新的迁移脚本，使用其 revision 值

Q: 迁移失败如何处理？

A:

1. 查看错误信息
2. 手动修复数据库状态
3. 修改迁移脚本
4. 重新执行迁移