xsl/xsl_node

Fork 0

Files

T

Your Name 40d2f3f6ac save code

2026-01-26 12:29:56 +08:00

21 KiB

Raw Blame History

后端数据库迁移文档

文档信息

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

1. 数据库迁移概述

1.1 什么是数据库迁移

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

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

1.2 为什么使用数据库迁移

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

2. 迁移工具

2.1 Alembic

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

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

2.2 安装 Alembic

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

# 使用 Poetry
poetry install

# 使用 pip
pip install -r requirements.txt

3. 迁移配置

3.1 初始化 Alembic

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

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

3.2 配置文件

Alembic 的主要配置文件是 alembic.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，用于配置数据库连接和迁移行为：

# 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 目录：

"""${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 生成迁移脚本

自动生成迁移脚本

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

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

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

手动创建迁移脚本

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

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

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

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

4.2 应用迁移

应用所有迁移

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

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

应用特定迁移

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

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

# 示例
alembic upgrade 1234abcd

4.3 回滚迁移

回滚到上一个版本

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

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

回滚到特定版本

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

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

# 示例
alembic downgrade 1234abcd

回滚到初始状态

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

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

4.4 查看迁移状态

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

# 在项目根目录执行
alembic current

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

# 在项目根目录执行
alembic history

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

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

5. 迁移最佳实践

5.1 迁移脚本管理

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

5.2 迁移脚本编写

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

5.3 常见迁移操作

添加表

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')

添加列

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

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

修改列

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
    )

添加索引

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 初始迁移

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

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

6.2 数据种子

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

# 在 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()

执行数据种子脚本：

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

7. 迁移示例

7.1 示例 1: 创建用户表

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

"""创建用户表

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

"""创建项目表

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 迁移失败

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

解决方法：

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

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

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

解决方法：

检查模型定义是否正确
手动编辑生成的迁移脚本
确保所有模型都已导入到 env.py 中

8.3 数据库连接错误

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

解决方法：

检查数据库服务是否运行
检查数据库连接字符串是否正确
检查数据库用户权限是否正确

8.4 迁移版本冲突

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

解决方法：

使用唯一的迁移版本号
在提交迁移脚本前检查版本冲突
如果发生冲突，重新生成迁移脚本

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 迁移最佳实践总结

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

9.3 变更记录

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

21 KiB Raw Blame History Unescape Escape

后端数据库迁移文档

文档信息

1. 数据库迁移概述

1.1 什么是数据库迁移

1.2 为什么使用数据库迁移

2. 迁移工具

2.1 Alembic

2.2 安装 Alembic

3. 迁移配置

3.1 初始化 Alembic

3.2 配置文件

3.3 环境配置

3.4 脚本模板

4. 迁移操作

4.1 生成迁移脚本

自动生成迁移脚本

手动创建迁移脚本

4.2 应用迁移

应用所有迁移

应用特定迁移

4.3 回滚迁移

回滚到上一个版本

回滚到特定版本

回滚到初始状态

4.4 查看迁移状态

5. 迁移最佳实践

5.1 迁移脚本管理

5.2 迁移脚本编写

5.3 常见迁移操作

添加表

添加列

修改列

添加索引

6. 数据库初始化

6.1 初始迁移

6.2 数据种子

7. 迁移示例

7.1 示例 1: 创建用户表

7.2 示例 2: 创建项目表

8. 常见问题

8.1 迁移失败

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

8.3 数据库连接错误

8.4 迁移版本冲突

9. 附录

9.1 常用命令汇总

9.2 迁移最佳实践总结

9.3 变更记录

21 KiB

Raw Blame History