# 后端数据库迁移文档

## 文档信息
- **文档版本**: V1.0
- **创建日期**: 2026-01-26
- **文档类型**: 后端数据库迁移文档

---

## 1. 数据库迁移概述

### 1.1 什么是数据库迁移

数据库迁移是一种管理数据库模式变更的方法，它允许您：

- 版本化数据库模式
- 跟踪数据库变更历史
- 在不同环境之间一致地应用数据库变更
- 回滚错误的数据库变更
- 协作开发数据库模式

### 1.2 为什么使用数据库迁移

- **可追溯性**: 记录所有数据库变更的历史
- **一致性**: 在开发、测试和生产环境中应用相同的变更
- **安全性**: 提供回滚机制，防止错误变更导致的数据丢失
- **协作**: 团队成员可以共享和审查数据库变更
- **自动化**: 简化数据库部署流程

---

## 2. 迁移工具

### 2.1 Alembic

本项目使用 **Alembic** 作为数据库迁移工具。Alembic 是 SQLAlchemy 的官方数据库迁移工具，提供了以下功能：

- 自动生成迁移脚本
- 管理迁移版本
- 应用和回滚迁移
- 支持多种数据库
- 与 SQLAlchemy 无缝集成

### 2.2 安装 Alembic

Alembic 已经在项目依赖中包含，使用以下命令安装：

```bash
# 使用 Poetry
poetry install

# 使用 pip
pip install -r requirements.txt
```

---

## 3. 迁移配置

### 3.1 初始化 Alembic

如果项目还没有初始化 Alembic，使用以下命令：

```bash
# 在项目根目录执行
alembic init alembic
```

### 3.2 配置文件

Alembic 的主要配置文件是 `alembic.ini`，位于项目根目录：

```ini
# alembic.ini

# A generic, single database configuration.

[alembic]
# path to migration scripts
script_location = alembic

# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# file_template = %%(year)d%%(month).2d%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s

# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .

# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python-dateutil library
# timezone =

# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40

# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false

# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false

# version location specification; This defaults
# to alembic/versions.  When using multiple version
# directories, initial revisions must be specified with --version-path.
# The path separator used here should be the separator specified by "version_path_separator" below.
# version_locations = %(here)s/bar:%(here)s/bat:alembic/versions

# version path separator; As mentioned above, this is the character used to split
# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
# Valid values for version_path_separator are:
#
# version_path_separator = :
# version_path_separator = ;
# version_path_separator = space
version_path_separator = os  # Use os.pathsep.
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8

sqlalchemy.url = mysql+pymysql://username:password@localhost:3306/project_management


[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts.  See the documentation for further
# detail and examples

# format using "black" - use the console_scripts runner, against the "black" entrypoint
# hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME

# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
```

### 3.3 环境配置

Alembic 的环境配置文件是 `alembic/env.py`，用于配置数据库连接和迁移行为：

```python
# alembic/env.py

from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
import os
import sys
from pathlib import Path

# 将项目根目录添加到 Python 路径
sys.path.append(str(Path(__file__).parent.parent))

# 导入模型和配置
from app.config import settings
from app.database.base import Base
from app.models import *  # 导入所有模型

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# 使用环境变量中的数据库 URL
config.set_main_option('sqlalchemy.url', settings.database_url)

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = Base.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()
```

### 3.4 脚本模板

Alembic 使用 `script.py.mako` 作为迁移脚本的模板，位于 `alembic` 目录：

```mako
"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}

# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}


def upgrade() -> None:
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    ${downgrades if downgrades else "pass"}
```

---

## 4. 迁移操作

### 4.1 生成迁移脚本

#### 自动生成迁移脚本

当您修改了模型定义后，使用以下命令自动生成迁移脚本：

```bash
# 在项目根目录执行
alembic revision --autogenerate -m "描述迁移内容"

# 示例
alembic revision --autogenerate -m "创建用户表"
```

#### 手动创建迁移脚本

如果需要手动创建迁移脚本，使用以下命令：

```bash
# 在项目根目录执行
alembic revision -m "描述迁移内容"

# 示例
alembic revision -m "添加用户状态字段"
```

然后编辑生成的迁移脚本，添加具体的迁移操作。

### 4.2 应用迁移

#### 应用所有迁移

使用以下命令将所有未应用的迁移应用到数据库：

```bash
# 在项目根目录执行
alembic upgrade head
```

#### 应用特定迁移

使用以下命令将迁移应用到特定版本：

```bash
# 在项目根目录执行
alembic upgrade <revision_id>

# 示例
alembic upgrade 1234abcd
```

### 4.3 回滚迁移

#### 回滚到上一个版本

使用以下命令回滚到上一个迁移版本：

```bash
# 在项目根目录执行
alembic downgrade -1
```

#### 回滚到特定版本

使用以下命令回滚到特定迁移版本：

```bash
# 在项目根目录执行
alembic downgrade <revision_id>

# 示例
alembic downgrade 1234abcd
```

#### 回滚到初始状态

使用以下命令回滚到初始状态：

```bash
# 在项目根目录执行
alembic downgrade base
```

### 4.4 查看迁移状态

使用以下命令查看当前的迁移状态：

```bash
# 在项目根目录执行
alembic current
```

使用以下命令查看所有迁移版本：

```bash
# 在项目根目录执行
alembic history

# 显示更详细的信息
alembic history --verbose

# 显示图形化的迁移树
alembic history --graph
```

---

## 5. 迁移最佳实践

### 5.1 迁移脚本管理

1. **清晰的迁移消息**: 使用描述性的迁移消息，说明迁移的目的
2. **版本控制**: 将迁移脚本纳入版本控制
3. **测试迁移**: 在应用到生产环境之前，在测试环境中测试迁移
4. **备份数据**: 在应用迁移之前，备份数据库
5. **逐步迁移**: 对于大型迁移，考虑分步骤进行

### 5.2 迁移脚本编写

1. **保持迁移脚本简单**: 每个迁移脚本只包含一个逻辑变更
2. **处理默认值**: 为新添加的列提供合理的默认值
3. **处理数据迁移**: 如果需要迁移数据，在迁移脚本中添加相应的代码
4. **处理约束**: 注意外键约束和唯一约束的处理
5. **编写回滚脚本**: 确保每个迁移都有对应的回滚操作

### 5.3 常见迁移操作

#### 添加表

```python
def upgrade() -> None:
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(length=50), nullable=False),
        sa.Column('password', sa.String(length=128), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('username')
    )

def downgrade() -> None:
    op.drop_table('users')
```

#### 添加列

```python
def upgrade() -> None:
    op.add_column(
        'users',
        sa.Column('email', sa.String(length=100), nullable=True)
    )

def downgrade() -> None:
    op.drop_column('users', 'email')
```

#### 修改列

```python
def upgrade() -> None:
    op.alter_column(
        'users',
        'email',
        existing_type=sa.String(length=100),
        nullable=False
    )

def downgrade() -> None:
    op.alter_column(
        'users',
        'email',
        existing_type=sa.String(length=100),
        nullable=True
    )
```

#### 添加索引

```python
def upgrade() -> None:
    op.create_index(
        op.f('ix_users_email'),
        'users',
        ['email'],
        unique=True
    )

def downgrade() -> None:
    op.drop_index(op.f('ix_users_email'), table_name='users')
```

---

## 6. 数据库初始化

### 6.1 初始迁移

当创建新项目时，需要执行初始迁移：

1. **创建模型**: 定义所有数据库模型
2. **生成初始迁移**: `alembic revision --autogenerate -m "初始迁移"`
3. **应用初始迁移**: `alembic upgrade head`

### 6.2 数据种子

在应用初始迁移后，可能需要添加一些初始数据，例如管理员用户：

```python
# 在 app/database/seed.py 中

from sqlalchemy.orm import Session
from app.models.user import User
from app.database.session import get_db
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def seed_data(db: Session) -> None:
    """添加初始数据"""
    # 检查是否已有管理员用户
    admin_user = db.query(User).filter(User.username == "admin").first()
    if not admin_user:
        # 创建管理员用户
        admin_user = User(
            username="admin",
            password=pwd_context.hash("123456"),
            real_name="系统管理员",
            department="ADMIN",
            role="ADMIN",
            status=1
        )
        db.add(admin_user)
    
    # 检查是否已有市场部用户
    marketing_user = db.query(User).filter(User.username == "marketing").first()
    if not marketing_user:
        # 创建市场部用户
        marketing_user = User(
            username="marketing",
            password=pwd_context.hash("123456"),
            real_name="市场部经理",
            department="MARKETING",
            role="MARKETING",
            status=1
        )
        db.add(marketing_user)
    
    # 检查是否已有其他部门用户
    other_user = db.query(User).filter(User.username == "other").first()
    if not other_user:
        # 创建其他部门用户
        other_user = User(
            username="other",
            password=pwd_context.hash("123456"),
            real_name="技术部经理",
            department="TECHNOLOGY",
            role="OTHER",
            status=1
        )
        db.add(other_user)
    
    db.commit()

if __name__ == "__main__":
    db = next(get_db())
    try:
        seed_data(db)
        print("数据种子添加成功")
    finally:
        db.close()
```

执行数据种子脚本：

```bash
# 在项目根目录执行
python -m app.database.seed
```

---

## 7. 迁移示例

### 7.1 示例 1: 创建用户表

**迁移脚本**: `alembic/versions/1234abcd_create_user_table.py`

```python
"""创建用户表

Revision ID: 1234abcd
Revises: 
Create Date: 2025-01-01 00:00:00.000000

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '1234abcd'
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table('sys_user',
    sa.Column('user_id', sa.String(length=32), nullable=False),
    sa.Column('username', sa.String(length=50), nullable=False),
    sa.Column('password', sa.String(length=128), nullable=False),
    sa.Column('real_name', sa.String(length=50), nullable=False),
    sa.Column('department', sa.String(length=50), nullable=False),
    sa.Column('phone', sa.String(length=20), nullable=True),
    sa.Column('email', sa.String(length=100), nullable=True),
    sa.Column('role', sa.String(length=20), nullable=False),
    sa.Column('status', sa.Integer(), nullable=False),
    sa.Column('create_time', sa.DateTime(), nullable=False),
    sa.Column('update_time', sa.DateTime(), nullable=False),
    sa.Column('last_login_time', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('user_id'),
    sa.UniqueConstraint('username')
    )
    op.create_index(op.f('ix_sys_user_department'), 'sys_user', ['department'], unique=False)
    op.create_index(op.f('ix_sys_user_role'), 'sys_user', ['role'], unique=False)
    op.create_index(op.f('ix_sys_user_status'), 'sys_user', ['status'], unique=False)


def downgrade() -> None:
    op.drop_index(op.f('ix_sys_user_status'), table_name='sys_user')
    op.drop_index(op.f('ix_sys_user_role'), table_name='sys_user')
    op.drop_index(op.f('ix_sys_user_department'), table_name='sys_user')
    op.drop_table('sys_user')
```

### 7.2 示例 2: 创建项目表

**迁移脚本**: `alembic/versions/5678efgh_create_project_table.py`

```python
"""创建项目表

Revision ID: 5678efgh
Revises: 1234abcd
Create Date: 2025-01-02 00:00:00.000000

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '5678efgh'
down_revision = '1234abcd'
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table('project',
    sa.Column('project_id', sa.String(length=32), nullable=False),
    sa.Column('project_no', sa.String(length=20), nullable=False),
    sa.Column('project_name', sa.String(length=200), nullable=False),
    sa.Column('status', sa.String(length=20), nullable=False),
    sa.Column('create_time', sa.DateTime(), nullable=False),
    sa.Column('creator', sa.String(length=50), nullable=False),
    sa.Column('update_time', sa.DateTime(), nullable=False),
    sa.Column('last_modifier', sa.String(length=50), nullable=True),
    sa.Column('leader', sa.String(length=50), nullable=False),
    sa.Column('phone', sa.String(length=20), nullable=True),
    sa.Column('email', sa.String(length=100), nullable=True),
    sa.Column('background', sa.Text(), nullable=True),
    sa.Column('goal', sa.Text(), nullable=True),
    sa.Column('scope', sa.Text(), nullable=True),
    sa.Column('start_date', sa.Date(), nullable=False),
    sa.Column('planned_end_date', sa.Date(), nullable=False),
    sa.Column('actual_end_date', sa.Date(), nullable=True),
    sa.Column('total_budget', sa.Numeric(precision=15, scale=2), nullable=False),
    sa.Column('used_budget', sa.Numeric(precision=15, scale=2), nullable=False),
    sa.Column('remaining_budget', sa.Numeric(precision=15, scale=2), nullable=False),
    sa.Column('remarks', sa.Text(), nullable=True),
    sa.PrimaryKeyConstraint('project_id'),
    sa.UniqueConstraint('project_no')
    )
    op.create_index(op.f('ix_project_create_time'), 'project', ['create_time'], unique=False)
    op.create_index(op.f('ix_project_creator'), 'project', ['creator'], unique=False)
    op.create_index(op.f('ix_project_leader'), 'project', ['leader'], unique=False)
    op.create_index(op.f('ix_project_status'), 'project', ['status'], unique=False)
    op.create_index(op.f('ix_project_update_time'), 'project', ['update_time'], unique=False)


def downgrade() -> None:
    op.drop_index(op.f('ix_project_update_time'), table_name='project')
    op.drop_index(op.f('ix_project_status'), table_name='project')
    op.drop_index(op.f('ix_project_leader'), table_name='project')
    op.drop_index(op.f('ix_project_creator'), table_name='project')
    op.drop_index(op.f('ix_project_create_time'), table_name='project')
    op.drop_table('project')
```

---

## 8. 常见问题

### 8.1 迁移失败

**问题**：迁移应用失败，出现错误

**解决方法**：
1. 查看错误信息，了解失败原因
2. 检查迁移脚本是否正确
3. 检查数据库连接是否正常
4. 如果需要，回滚到之前的版本并修复问题

### 8.2 自动生成的迁移脚本不正确

**问题**：Alembic 自动生成的迁移脚本与预期不符

**解决方法**：
1. 检查模型定义是否正确
2. 手动编辑生成的迁移脚本
3. 确保所有模型都已导入到 `env.py` 中

### 8.3 数据库连接错误

**问题**：无法连接到数据库进行迁移

**解决方法**：
1. 检查数据库服务是否运行
2. 检查数据库连接字符串是否正确
3. 检查数据库用户权限是否正确

### 8.4 迁移版本冲突

**问题**：多个开发者创建了相同版本号的迁移

**解决方法**：
1. 使用唯一的迁移版本号
2. 在提交迁移脚本前检查版本冲突
3. 如果发生冲突，重新生成迁移脚本

---

## 9. 附录

### 9.1 常用命令汇总

| 命令 | 说明 | 示例 |
|------|------|------|
| `alembic init` | 初始化 Alembic | `alembic init alembic` |
| `alembic revision --autogenerate` | 自动生成迁移脚本 | `alembic revision --autogenerate -m "创建用户表"` |
| `alembic revision` | 手动创建迁移脚本 | `alembic revision -m "添加用户状态字段"` |
| `alembic upgrade head` | 应用所有迁移 | `alembic upgrade head` |
| `alembic upgrade <revision>` | 应用到特定版本 | `alembic upgrade 1234abcd` |
| `alembic downgrade -1` | 回滚到上一个版本 | `alembic downgrade -1` |
| `alembic downgrade <revision>` | 回滚到特定版本 | `alembic downgrade 1234abcd` |
| `alembic downgrade base` | 回滚到初始状态 | `alembic downgrade base` |
| `alembic current` | 查看当前版本 | `alembic current` |
| `alembic history` | 查看所有版本 | `alembic history` |
| `alembic history --verbose` | 查看详细版本信息 | `alembic history --verbose` |
| `alembic history --graph` | 查看迁移树 | `alembic history --graph` |

### 9.2 迁移最佳实践总结

1. **版本控制**: 将迁移脚本纳入版本控制
2. **测试**: 在应用到生产环境之前测试迁移
3. **备份**: 在应用迁移之前备份数据库
4. **简单**: 每个迁移只包含一个逻辑变更
5. **清晰**: 使用描述性的迁移消息
6. **完整**: 编写正确的升级和降级脚本
7. **协作**: 与团队成员协调迁移工作
8. **监控**: 监控迁移执行情况

### 9.3 变更记录

| 版本 | 日期 | 修改人 | 修改内容 |
|------|------|--------|----------|
| V1.0 | 2026-01-26 | - | 初始版本创建 |